mirror of
https://github.com/python/cpython
synced 2026-11-14 15:47:43 +03:00
fd83a82c14
gh-159875: Keep the SSL protocol of a closed transport after connection loss (GH-160069)
* gh-159875: Add tests for an SSL transport closed after connection loss
An application that closes its transport from connection_lost(), as aiohttp
does, must be left with a transport whose write(), get_write_buffer_size()
and set_protocol() behave as after closing a live connection, and
loop.sendfile()'s fallback must then raise ConnectionError rather than
AttributeError from set_protocol().
The sendfile test reads the file synchronously, so that the fallback is
waiting in drain() when the connection is lost; a read in the executor that
completes in the same loop iteration as the reset lets the fallback finish
first and hides the bug.
test_create_connection_memory_leak now keeps the transport past its check,
as its comment says it covers, so that it would see an SSLContext kept alive
by a transport the application still holds.
* gh-159875: Keep the SSL protocol of a closed transport after connection loss
_SSLProtocolTransport.close() dropped its _ssl_protocol when called a second
time, and the transport is already marked closed once the connection is lost,
so an application closing it from connection_lost() left a transport whose
write(), set_protocol(), get_write_buffer_size(), ... raise AttributeError.
That branch was also what released the SSLContext of a transport the
application keeps. Keep the protocol, and release the TLS state from
SSLProtocol.connection_lost() instead: one loop iteration later when the
application's connection_lost() was scheduled, so that it can still read
get_extra_info('ssl_object') and ('sslcontext').
---------
(cherry picked from commit 0626b14d31)
Co-authored-by: Mauricio Paez <jose.m.paez@icloud.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Andrew Svetlov <andrew.svetlov@gmail.com>
1086 lines
39 KiB
Python
1086 lines
39 KiB
Python
"""Tests for asyncio/sslproto.py."""
|
|
|
|
import logging
|
|
import socket
|
|
import struct
|
|
import unittest
|
|
import weakref
|
|
from test import support
|
|
from test.support import socket_helper
|
|
from unittest import mock
|
|
try:
|
|
import ssl
|
|
except ImportError:
|
|
ssl = None
|
|
|
|
import asyncio
|
|
from asyncio import log
|
|
from asyncio import protocols
|
|
from asyncio import sslproto
|
|
from test.test_asyncio import utils as test_utils
|
|
from test.test_asyncio import functional as func_tests
|
|
|
|
|
|
def tearDownModule():
|
|
asyncio.events._set_event_loop_policy(None)
|
|
|
|
|
|
@unittest.skipIf(ssl is None, 'No ssl module')
|
|
class SslProtoHandshakeTests(test_utils.TestCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.loop = asyncio.new_event_loop()
|
|
self.set_event_loop(self.loop)
|
|
|
|
def ssl_protocol(self, *, waiter=None, proto=None):
|
|
sslcontext = test_utils.dummy_ssl_context()
|
|
if proto is None: # app protocol
|
|
proto = asyncio.Protocol()
|
|
ssl_proto = sslproto.SSLProtocol(self.loop, proto, sslcontext, waiter,
|
|
ssl_handshake_timeout=0.1)
|
|
self.assertIs(ssl_proto._app_transport.get_protocol(), proto)
|
|
self.addCleanup(ssl_proto._app_transport.close)
|
|
return ssl_proto
|
|
|
|
def connection_made(self, ssl_proto, *, do_handshake=None):
|
|
transport = mock.Mock()
|
|
sslobj = mock.Mock()
|
|
# emulate reading decompressed data
|
|
sslobj.read.side_effect = ssl.SSLWantReadError
|
|
sslobj.write.side_effect = ssl.SSLWantReadError
|
|
if do_handshake is not None:
|
|
sslobj.do_handshake = do_handshake
|
|
ssl_proto._sslobj = sslobj
|
|
ssl_proto.connection_made(transport)
|
|
return transport
|
|
|
|
def test_handshake_timeout_zero(self):
|
|
sslcontext = test_utils.dummy_ssl_context()
|
|
app_proto = mock.Mock()
|
|
waiter = mock.Mock()
|
|
with self.assertRaisesRegex(ValueError, 'a positive number'):
|
|
sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter,
|
|
ssl_handshake_timeout=0)
|
|
|
|
def test_handshake_timeout_negative(self):
|
|
sslcontext = test_utils.dummy_ssl_context()
|
|
app_proto = mock.Mock()
|
|
waiter = mock.Mock()
|
|
with self.assertRaisesRegex(ValueError, 'a positive number'):
|
|
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)
|
|
self.connection_made(
|
|
ssl_proto,
|
|
do_handshake=mock.Mock(side_effect=ssl.SSLWantReadError)
|
|
)
|
|
ssl_proto.eof_received()
|
|
test_utils.run_briefly(self.loop)
|
|
self.assertIsInstance(waiter.exception(), ConnectionResetError)
|
|
|
|
def test_fatal_error_no_name_error(self):
|
|
# From issue #363.
|
|
# _fatal_error() generates a NameError if sslproto.py
|
|
# does not import base_events.
|
|
waiter = self.loop.create_future()
|
|
ssl_proto = self.ssl_protocol(waiter=waiter)
|
|
# Temporarily turn off error logging so as not to spoil test output.
|
|
log_level = log.logger.getEffectiveLevel()
|
|
log.logger.setLevel(logging.FATAL)
|
|
try:
|
|
ssl_proto._fatal_error(None)
|
|
finally:
|
|
# Restore error logging.
|
|
log.logger.setLevel(log_level)
|
|
|
|
def test_connection_lost(self):
|
|
# From issue #472.
|
|
# yield from waiter hang if lost_connection was called.
|
|
waiter = self.loop.create_future()
|
|
ssl_proto = self.ssl_protocol(waiter=waiter)
|
|
self.connection_made(
|
|
ssl_proto,
|
|
do_handshake=mock.Mock(side_effect=ssl.SSLWantReadError)
|
|
)
|
|
ssl_proto.connection_lost(ConnectionAbortedError)
|
|
test_utils.run_briefly(self.loop)
|
|
self.assertIsInstance(waiter.exception(), ConnectionAbortedError)
|
|
|
|
def test_connection_lost_during_handshake(self):
|
|
# The app protocol gets connection_lost() if and only if it got
|
|
# connection_made(). With start_tls() (call_connection_made=False)
|
|
# it is already connected when the handshake starts.
|
|
for call_connection_made in (True, False):
|
|
with self.subTest(call_connection_made=call_connection_made):
|
|
app_proto = mock.Mock(spec=asyncio.Protocol)
|
|
ssl_proto = sslproto.SSLProtocol(
|
|
self.loop, app_proto, test_utils.dummy_ssl_context(),
|
|
waiter=None, call_connection_made=call_connection_made)
|
|
self.connection_made(
|
|
ssl_proto,
|
|
do_handshake=mock.Mock(side_effect=ssl.SSLWantReadError)
|
|
)
|
|
exc = ConnectionResetError()
|
|
ssl_proto.connection_lost(exc)
|
|
test_utils.run_briefly(self.loop)
|
|
app_proto.connection_made.assert_not_called()
|
|
if call_connection_made:
|
|
app_proto.connection_lost.assert_not_called()
|
|
else:
|
|
app_proto.connection_lost.assert_called_once_with(exc)
|
|
|
|
def test_connection_lost_when_busy(self):
|
|
# gh-118950: SSLProtocol.connection_lost not being called when OSError
|
|
# is thrown on asyncio.write.
|
|
sock = mock.Mock()
|
|
sock.fileno = mock.Mock(return_value=12345)
|
|
sock.send = mock.Mock(side_effect=BrokenPipeError)
|
|
|
|
# construct StreamWriter chain that contains loop dependant logic this emulates
|
|
# what _make_ssl_transport() does in BaseSelectorEventLoop
|
|
reader = asyncio.StreamReader(limit=2 ** 16, loop=self.loop)
|
|
protocol = asyncio.StreamReaderProtocol(reader, loop=self.loop)
|
|
ssl_proto = self.ssl_protocol(proto=protocol)
|
|
|
|
# emulate reading decompressed data
|
|
sslobj = mock.Mock()
|
|
sslobj.read.side_effect = ssl.SSLWantReadError
|
|
sslobj.write.side_effect = ssl.SSLWantReadError
|
|
ssl_proto._sslobj = sslobj
|
|
|
|
# emulate outgoing data
|
|
data = b'An interesting message'
|
|
|
|
outgoing = mock.Mock()
|
|
outgoing.read = mock.Mock(return_value=data)
|
|
outgoing.pending = len(data)
|
|
ssl_proto._outgoing = outgoing
|
|
|
|
# use correct socket transport to initialize the SSLProtocol
|
|
self.loop._make_socket_transport(sock, ssl_proto)
|
|
|
|
transport = ssl_proto._app_transport
|
|
writer = asyncio.StreamWriter(transport, protocol, reader, self.loop)
|
|
|
|
async def main():
|
|
# writes data to transport
|
|
async def write():
|
|
writer.write(data)
|
|
await writer.drain()
|
|
|
|
# try to write for the first time
|
|
await write()
|
|
# try to write for the second time, this raises as the connection_lost
|
|
# callback should be done with error
|
|
with self.assertRaises(ConnectionResetError):
|
|
await write()
|
|
|
|
self.loop.run_until_complete(main())
|
|
|
|
def test_close_during_handshake(self):
|
|
# bpo-29743 Closing transport during handshake process leaks socket
|
|
waiter = self.loop.create_future()
|
|
ssl_proto = self.ssl_protocol(waiter=waiter)
|
|
|
|
transport = self.connection_made(
|
|
ssl_proto,
|
|
do_handshake=mock.Mock(side_effect=ssl.SSLWantReadError)
|
|
)
|
|
test_utils.run_briefly(self.loop)
|
|
|
|
ssl_proto._app_transport.close()
|
|
self.assertTrue(transport._force_close.called)
|
|
|
|
def test_close_during_ssl_over_ssl(self):
|
|
# gh-113214: passing exceptions from the inner wrapped SSL protocol to the
|
|
# shim transport provided by the outer SSL protocol should not raise
|
|
# attribute errors
|
|
outer = self.ssl_protocol(proto=self.ssl_protocol())
|
|
self.connection_made(outer)
|
|
# Closing the outer app transport should not raise an exception
|
|
messages = []
|
|
self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx))
|
|
outer._app_transport.close()
|
|
self.assertEqual(messages, [])
|
|
|
|
def test_get_extra_info_on_closed_connection(self):
|
|
waiter = self.loop.create_future()
|
|
ssl_proto = self.ssl_protocol(waiter=waiter)
|
|
self.assertIsNone(ssl_proto._get_extra_info('socket'))
|
|
default = object()
|
|
self.assertIs(ssl_proto._get_extra_info('socket', default), default)
|
|
self.connection_made(ssl_proto)
|
|
self.assertIsNotNone(ssl_proto._get_extra_info('socket'))
|
|
ssl_proto.connection_lost(None)
|
|
self.assertIsNone(ssl_proto._get_extra_info('socket'))
|
|
|
|
def test_set_new_app_protocol(self):
|
|
waiter = self.loop.create_future()
|
|
ssl_proto = self.ssl_protocol(waiter=waiter)
|
|
new_app_proto = asyncio.Protocol()
|
|
ssl_proto._app_transport.set_protocol(new_app_proto)
|
|
self.assertIs(ssl_proto._app_transport.get_protocol(), new_app_proto)
|
|
self.assertIs(ssl_proto._app_protocol, new_app_proto)
|
|
|
|
def test_data_received_after_closing(self):
|
|
ssl_proto = self.ssl_protocol()
|
|
self.connection_made(ssl_proto)
|
|
transp = ssl_proto._app_transport
|
|
|
|
transp.close()
|
|
|
|
# should not raise
|
|
self.assertIsNone(ssl_proto.buffer_updated(5))
|
|
|
|
def test_write_after_closing(self):
|
|
ssl_proto = self.ssl_protocol()
|
|
self.connection_made(ssl_proto)
|
|
transp = ssl_proto._app_transport
|
|
transp.close()
|
|
|
|
# should not raise
|
|
self.assertIsNone(transp.write(b'data'))
|
|
|
|
def test_transport_usable_after_connection_lost_and_close(self):
|
|
# connection_lost() marks the app transport closed, so the
|
|
# application's own close() afterwards (often made from its
|
|
# protocol's connection_lost()) must not be taken as a second
|
|
# close() that leaves the transport without its SSL protocol.
|
|
ssl_proto = self.ssl_protocol()
|
|
self.connection_made(ssl_proto)
|
|
transp = ssl_proto._app_transport
|
|
ssl_proto.connection_lost(None)
|
|
transp.close()
|
|
|
|
# should not raise, as after closing a live connection
|
|
self.assertIsNone(transp.write(b'data'))
|
|
self.assertEqual(transp.get_write_buffer_size(), 0)
|
|
new_proto = asyncio.Protocol()
|
|
transp.set_protocol(new_proto) # as loop.sendfile()'s fallback does
|
|
self.assertIs(transp.get_protocol(), new_proto)
|
|
transp.abort()
|
|
self.assertTrue(transp.is_closing())
|
|
|
|
def test_pause_reading_if_ssl_replies_are_not_read(self):
|
|
# Output which OpenSSL produces on its own while reading (alerts,
|
|
# KeyUpdate, ...) must not pile up if the peer does not read it.
|
|
ssl_proto = self.ssl_protocol()
|
|
transport = self.connection_made(ssl_proto, do_handshake=mock.Mock())
|
|
reply = b'x' * (ssl_proto._incoming_high_water // 2)
|
|
|
|
def read(*args):
|
|
ssl_proto._outgoing.write(reply)
|
|
raise ssl.SSLWantReadError
|
|
ssl_proto._sslobj.read.side_effect = read
|
|
|
|
ssl_proto.pause_writing()
|
|
ssl_proto.buffer_updated(0)
|
|
transport.pause_reading.assert_not_called()
|
|
ssl_proto.buffer_updated(0)
|
|
transport.pause_reading.assert_called_once_with()
|
|
transport.resume_reading.assert_not_called()
|
|
transport.write.assert_not_called()
|
|
|
|
ssl_proto.resume_writing()
|
|
transport.write.assert_called_once_with(reply * 2)
|
|
transport.resume_reading.assert_called_once_with()
|
|
|
|
def test_no_pause_reading_if_app_data_is_not_read(self):
|
|
# Peers which both write a lot before they read must not deadlock.
|
|
ssl_proto = self.ssl_protocol()
|
|
transport = self.connection_made(ssl_proto, do_handshake=mock.Mock())
|
|
ssl_proto._sslobj.write.side_effect = ssl_proto._outgoing.write
|
|
data = b'x' * ssl_proto._incoming_high_water
|
|
|
|
ssl_proto.pause_writing()
|
|
ssl_proto._app_transport.write(data)
|
|
ssl_proto.buffer_updated(0)
|
|
transport.pause_reading.assert_not_called()
|
|
|
|
ssl_proto.resume_writing()
|
|
transport.write.assert_called_once_with(data)
|
|
transport.resume_reading.assert_not_called()
|
|
|
|
|
|
##############################################################################
|
|
# Start TLS Tests
|
|
##############################################################################
|
|
|
|
|
|
class BaseStartTLS(func_tests.FunctionalTestCaseMixin):
|
|
|
|
PAYLOAD_SIZE = 1024 * 100
|
|
TIMEOUT = support.LONG_TIMEOUT
|
|
|
|
def new_loop(self):
|
|
raise NotImplementedError
|
|
|
|
def test_buf_feed_data(self):
|
|
|
|
class Proto(asyncio.BufferedProtocol):
|
|
|
|
def __init__(self, bufsize, usemv):
|
|
self.buf = bytearray(bufsize)
|
|
self.mv = memoryview(self.buf)
|
|
self.data = b''
|
|
self.usemv = usemv
|
|
|
|
def get_buffer(self, sizehint):
|
|
if self.usemv:
|
|
return self.mv
|
|
else:
|
|
return self.buf
|
|
|
|
def buffer_updated(self, nsize):
|
|
if self.usemv:
|
|
self.data += self.mv[:nsize]
|
|
else:
|
|
self.data += self.buf[:nsize]
|
|
|
|
for usemv in [False, True]:
|
|
proto = Proto(1, usemv)
|
|
protocols._feed_data_to_buffered_proto(proto, b'12345')
|
|
self.assertEqual(proto.data, b'12345')
|
|
|
|
proto = Proto(2, usemv)
|
|
protocols._feed_data_to_buffered_proto(proto, b'12345')
|
|
self.assertEqual(proto.data, b'12345')
|
|
|
|
proto = Proto(2, usemv)
|
|
protocols._feed_data_to_buffered_proto(proto, b'1234')
|
|
self.assertEqual(proto.data, b'1234')
|
|
|
|
proto = Proto(4, usemv)
|
|
protocols._feed_data_to_buffered_proto(proto, b'1234')
|
|
self.assertEqual(proto.data, b'1234')
|
|
|
|
proto = Proto(100, usemv)
|
|
protocols._feed_data_to_buffered_proto(proto, b'12345')
|
|
self.assertEqual(proto.data, b'12345')
|
|
|
|
proto = Proto(0, usemv)
|
|
with self.assertRaisesRegex(RuntimeError, 'empty buffer'):
|
|
protocols._feed_data_to_buffered_proto(proto, b'12345')
|
|
|
|
def test_start_tls_client_reg_proto_1(self):
|
|
HELLO_MSG = b'1' * self.PAYLOAD_SIZE
|
|
|
|
server_context = test_utils.simple_server_sslcontext()
|
|
client_context = test_utils.simple_client_sslcontext()
|
|
|
|
def serve(sock):
|
|
sock.settimeout(self.TIMEOUT)
|
|
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
sock.start_tls(server_context, server_side=True)
|
|
|
|
sock.sendall(b'O')
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
sock.shutdown(socket.SHUT_RDWR)
|
|
sock.close()
|
|
|
|
class ClientProto(asyncio.Protocol):
|
|
def __init__(self, on_data, on_eof):
|
|
self.on_data = on_data
|
|
self.on_eof = on_eof
|
|
self.con_made_cnt = 0
|
|
|
|
def connection_made(proto, tr):
|
|
proto.con_made_cnt += 1
|
|
# Ensure connection_made gets called only once.
|
|
self.assertEqual(proto.con_made_cnt, 1)
|
|
|
|
def data_received(self, data):
|
|
self.on_data.set_result(data)
|
|
|
|
def eof_received(self):
|
|
self.on_eof.set_result(True)
|
|
|
|
async def client(addr):
|
|
await asyncio.sleep(0.5)
|
|
|
|
on_data = self.loop.create_future()
|
|
on_eof = self.loop.create_future()
|
|
|
|
tr, proto = await self.loop.create_connection(
|
|
lambda: ClientProto(on_data, on_eof), *addr)
|
|
|
|
tr.write(HELLO_MSG)
|
|
new_tr = await self.loop.start_tls(tr, proto, client_context)
|
|
|
|
self.assertEqual(await on_data, b'O')
|
|
new_tr.write(HELLO_MSG)
|
|
await on_eof
|
|
|
|
new_tr.close()
|
|
|
|
with self.tcp_server(serve, timeout=self.TIMEOUT) as srv:
|
|
self.loop.run_until_complete(
|
|
asyncio.wait_for(client(srv.addr),
|
|
timeout=support.SHORT_TIMEOUT))
|
|
|
|
# No garbage is left if SSL is closed uncleanly
|
|
client_context = weakref.ref(client_context)
|
|
support.gc_collect()
|
|
self.assertIsNone(client_context())
|
|
|
|
def test_create_connection_memory_leak(self):
|
|
HELLO_MSG = b'1' * self.PAYLOAD_SIZE
|
|
|
|
server_context = test_utils.simple_server_sslcontext()
|
|
client_context = test_utils.simple_client_sslcontext()
|
|
|
|
def serve(sock):
|
|
sock.settimeout(self.TIMEOUT)
|
|
|
|
sock.start_tls(server_context, server_side=True)
|
|
|
|
sock.sendall(b'O')
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
sock.shutdown(socket.SHUT_RDWR)
|
|
sock.close()
|
|
|
|
class ClientProto(asyncio.Protocol):
|
|
def __init__(self, on_data, on_eof):
|
|
self.on_data = on_data
|
|
self.on_eof = on_eof
|
|
self.con_made_cnt = 0
|
|
|
|
def connection_made(proto, tr):
|
|
# XXX: We assume user stores the transport in protocol
|
|
proto.tr = tr
|
|
proto.con_made_cnt += 1
|
|
# Ensure connection_made gets called only once.
|
|
self.assertEqual(proto.con_made_cnt, 1)
|
|
|
|
def data_received(self, data):
|
|
self.on_data.set_result(data)
|
|
|
|
def eof_received(self):
|
|
self.on_eof.set_result(True)
|
|
|
|
async def client(addr):
|
|
await asyncio.sleep(0.5)
|
|
|
|
on_data = self.loop.create_future()
|
|
on_eof = self.loop.create_future()
|
|
|
|
tr, proto = await self.loop.create_connection(
|
|
lambda: ClientProto(on_data, on_eof), *addr,
|
|
ssl=client_context)
|
|
|
|
self.assertEqual(await on_data, b'O')
|
|
tr.write(HELLO_MSG)
|
|
await on_eof
|
|
|
|
tr.close()
|
|
|
|
with self.tcp_server(serve, timeout=self.TIMEOUT) as srv:
|
|
self.loop.run_until_complete(
|
|
asyncio.wait_for(client(srv.addr),
|
|
timeout=support.SHORT_TIMEOUT))
|
|
|
|
# No garbage is left for SSL client from loop.create_connection, even
|
|
# if user stores the SSLTransport in corresponding protocol instance
|
|
client_context = weakref.ref(client_context)
|
|
support.gc_collect()
|
|
self.assertIsNone(client_context())
|
|
|
|
@socket_helper.skip_if_tcp_blackhole
|
|
def test_start_tls_client_buf_proto_1(self):
|
|
HELLO_MSG = b'1' * self.PAYLOAD_SIZE
|
|
|
|
server_context = test_utils.simple_server_sslcontext()
|
|
client_context = test_utils.simple_client_sslcontext()
|
|
client_con_made_calls = 0
|
|
|
|
def serve(sock):
|
|
sock.settimeout(self.TIMEOUT)
|
|
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
sock.start_tls(server_context, server_side=True)
|
|
|
|
sock.sendall(b'O')
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
sock.sendall(b'2')
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
sock.shutdown(socket.SHUT_RDWR)
|
|
sock.close()
|
|
|
|
class ClientProtoFirst(asyncio.BufferedProtocol):
|
|
def __init__(self, on_data):
|
|
self.on_data = on_data
|
|
self.buf = bytearray(1)
|
|
|
|
def connection_made(self, tr):
|
|
nonlocal client_con_made_calls
|
|
client_con_made_calls += 1
|
|
|
|
def get_buffer(self, sizehint):
|
|
return self.buf
|
|
|
|
def buffer_updated(slf, nsize):
|
|
self.assertEqual(nsize, 1)
|
|
slf.on_data.set_result(bytes(slf.buf[:nsize]))
|
|
|
|
class ClientProtoSecond(asyncio.Protocol):
|
|
def __init__(self, on_data, on_eof):
|
|
self.on_data = on_data
|
|
self.on_eof = on_eof
|
|
self.con_made_cnt = 0
|
|
|
|
def connection_made(self, tr):
|
|
nonlocal client_con_made_calls
|
|
client_con_made_calls += 1
|
|
|
|
def data_received(self, data):
|
|
self.on_data.set_result(data)
|
|
|
|
def eof_received(self):
|
|
self.on_eof.set_result(True)
|
|
|
|
async def client(addr):
|
|
await asyncio.sleep(0.5)
|
|
|
|
on_data1 = self.loop.create_future()
|
|
on_data2 = self.loop.create_future()
|
|
on_eof = self.loop.create_future()
|
|
|
|
tr, proto = await self.loop.create_connection(
|
|
lambda: ClientProtoFirst(on_data1), *addr)
|
|
|
|
tr.write(HELLO_MSG)
|
|
new_tr = await self.loop.start_tls(tr, proto, client_context)
|
|
|
|
self.assertEqual(await on_data1, b'O')
|
|
new_tr.write(HELLO_MSG)
|
|
|
|
new_tr.set_protocol(ClientProtoSecond(on_data2, on_eof))
|
|
self.assertEqual(await on_data2, b'2')
|
|
new_tr.write(HELLO_MSG)
|
|
await on_eof
|
|
|
|
new_tr.close()
|
|
|
|
# connection_made() should be called only once -- when
|
|
# we establish connection for the first time. Start TLS
|
|
# doesn't call connection_made() on application protocols.
|
|
self.assertEqual(client_con_made_calls, 1)
|
|
|
|
with self.tcp_server(serve, timeout=self.TIMEOUT) as srv:
|
|
self.loop.run_until_complete(
|
|
asyncio.wait_for(client(srv.addr),
|
|
timeout=self.TIMEOUT))
|
|
|
|
def test_start_tls_slow_client_cancel(self):
|
|
HELLO_MSG = b'1' * self.PAYLOAD_SIZE
|
|
|
|
client_context = test_utils.simple_client_sslcontext()
|
|
server_waits_on_handshake = self.loop.create_future()
|
|
|
|
def serve(sock):
|
|
sock.settimeout(self.TIMEOUT)
|
|
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
try:
|
|
self.loop.call_soon_threadsafe(
|
|
server_waits_on_handshake.set_result, None)
|
|
data = sock.recv_all(1024 * 1024)
|
|
except ConnectionAbortedError:
|
|
pass
|
|
finally:
|
|
sock.close()
|
|
|
|
class ClientProto(asyncio.Protocol):
|
|
def __init__(self, on_data, on_eof, on_con_lost):
|
|
self.on_data = on_data
|
|
self.on_eof = on_eof
|
|
self.on_con_lost = on_con_lost
|
|
self.con_made_cnt = 0
|
|
|
|
def connection_made(proto, tr):
|
|
proto.con_made_cnt += 1
|
|
# Ensure connection_made gets called only once.
|
|
self.assertEqual(proto.con_made_cnt, 1)
|
|
|
|
def data_received(self, data):
|
|
self.on_data.set_result(data)
|
|
|
|
def eof_received(self):
|
|
self.on_eof.set_result(True)
|
|
|
|
def connection_lost(self, exc):
|
|
self.on_con_lost.set_result(exc)
|
|
|
|
async def client(addr):
|
|
await asyncio.sleep(0.5)
|
|
|
|
on_data = self.loop.create_future()
|
|
on_eof = self.loop.create_future()
|
|
on_con_lost = self.loop.create_future()
|
|
|
|
tr, proto = await self.loop.create_connection(
|
|
lambda: ClientProto(on_data, on_eof, on_con_lost), *addr)
|
|
|
|
tr.write(HELLO_MSG)
|
|
|
|
await server_waits_on_handshake
|
|
|
|
with self.assertRaises(asyncio.TimeoutError):
|
|
await asyncio.wait_for(
|
|
self.loop.start_tls(tr, proto, client_context),
|
|
0.5)
|
|
|
|
self.assertIsNone(
|
|
await asyncio.wait_for(on_con_lost, support.LOOPBACK_TIMEOUT))
|
|
|
|
with self.tcp_server(serve, timeout=self.TIMEOUT) as srv:
|
|
self.loop.run_until_complete(
|
|
asyncio.wait_for(client(srv.addr),
|
|
timeout=support.SHORT_TIMEOUT))
|
|
|
|
def _test_start_tls_streams_failed_handshake(self, serve, expected_exc,
|
|
**kwargs):
|
|
# wait_closed() must not hang if the connection is lost while
|
|
# the handshake started by start_tls() is still in progress.
|
|
client_context = test_utils.simple_client_sslcontext()
|
|
|
|
async def client(addr):
|
|
reader, writer = await asyncio.open_connection(*addr)
|
|
with self.assertRaises(expected_exc):
|
|
await writer.start_tls(client_context, **kwargs)
|
|
writer.close()
|
|
with self.assertRaises(expected_exc):
|
|
await writer.wait_closed()
|
|
|
|
with self.tcp_server(serve, timeout=self.TIMEOUT) as srv:
|
|
self.loop.run_until_complete(
|
|
asyncio.wait_for(client(srv.addr),
|
|
timeout=support.SHORT_TIMEOUT))
|
|
|
|
def test_start_tls_streams_handshake_reset(self):
|
|
def serve(sock):
|
|
sock.settimeout(self.TIMEOUT)
|
|
sock.recv(1024 * 1024) # ClientHello
|
|
# Send RST instead of FIN.
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
|
|
struct.pack('ii', 1, 0))
|
|
sock.close()
|
|
|
|
self._test_start_tls_streams_failed_handshake(
|
|
serve, ConnectionError)
|
|
|
|
def test_start_tls_streams_handshake_timeout(self):
|
|
def serve(sock):
|
|
sock.settimeout(self.TIMEOUT)
|
|
try:
|
|
while sock.recv(1024 * 1024):
|
|
pass
|
|
except ConnectionError:
|
|
pass
|
|
finally:
|
|
sock.close()
|
|
|
|
self._test_start_tls_streams_failed_handshake(
|
|
serve, ConnectionAbortedError, ssl_handshake_timeout=0.1)
|
|
|
|
@socket_helper.skip_if_tcp_blackhole
|
|
def test_start_tls_server_1(self):
|
|
HELLO_MSG = b'1' * self.PAYLOAD_SIZE
|
|
ANSWER = b'answer'
|
|
|
|
server_context = test_utils.simple_server_sslcontext()
|
|
client_context = test_utils.simple_client_sslcontext()
|
|
answer = None
|
|
|
|
def client(sock, addr):
|
|
nonlocal answer
|
|
sock.settimeout(self.TIMEOUT)
|
|
|
|
sock.connect(addr)
|
|
data = sock.recv_all(len(HELLO_MSG))
|
|
self.assertEqual(len(data), len(HELLO_MSG))
|
|
|
|
sock.start_tls(client_context)
|
|
sock.sendall(HELLO_MSG)
|
|
answer = sock.recv_all(len(ANSWER))
|
|
sock.close()
|
|
|
|
class ServerProto(asyncio.Protocol):
|
|
def __init__(self, on_con, on_con_lost, on_got_hello):
|
|
self.on_con = on_con
|
|
self.on_con_lost = on_con_lost
|
|
self.on_got_hello = on_got_hello
|
|
self.data = b''
|
|
self.transport = None
|
|
|
|
def connection_made(self, tr):
|
|
self.transport = tr
|
|
self.on_con.set_result(tr)
|
|
|
|
def replace_transport(self, tr):
|
|
self.transport = tr
|
|
|
|
def data_received(self, data):
|
|
self.data += data
|
|
if len(self.data) >= len(HELLO_MSG):
|
|
self.on_got_hello.set_result(None)
|
|
|
|
def connection_lost(self, exc):
|
|
self.transport = None
|
|
if exc is None:
|
|
self.on_con_lost.set_result(None)
|
|
else:
|
|
self.on_con_lost.set_exception(exc)
|
|
|
|
async def main(proto, on_con, on_con_lost, on_got_hello):
|
|
tr = await on_con
|
|
tr.write(HELLO_MSG)
|
|
|
|
self.assertEqual(proto.data, b'')
|
|
|
|
new_tr = await self.loop.start_tls(
|
|
tr, proto, server_context,
|
|
server_side=True,
|
|
ssl_handshake_timeout=self.TIMEOUT)
|
|
proto.replace_transport(new_tr)
|
|
|
|
await on_got_hello
|
|
new_tr.write(ANSWER)
|
|
|
|
await on_con_lost
|
|
self.assertEqual(proto.data, HELLO_MSG)
|
|
new_tr.close()
|
|
|
|
async def run_main():
|
|
on_con = self.loop.create_future()
|
|
on_con_lost = self.loop.create_future()
|
|
on_got_hello = self.loop.create_future()
|
|
proto = ServerProto(on_con, on_con_lost, on_got_hello)
|
|
|
|
server = await self.loop.create_server(
|
|
lambda: proto, '127.0.0.1', 0)
|
|
addr = server.sockets[0].getsockname()
|
|
|
|
with self.tcp_client(lambda sock: client(sock, addr),
|
|
timeout=self.TIMEOUT):
|
|
await asyncio.wait_for(
|
|
main(proto, on_con, on_con_lost, on_got_hello),
|
|
timeout=self.TIMEOUT)
|
|
|
|
server.close()
|
|
await server.wait_closed()
|
|
self.assertEqual(answer, ANSWER)
|
|
|
|
self.loop.run_until_complete(run_main())
|
|
|
|
def test_start_tls_wrong_args(self):
|
|
async def main():
|
|
with self.assertRaisesRegex(TypeError, 'SSLContext, got'):
|
|
await self.loop.start_tls(None, None, None)
|
|
|
|
sslctx = test_utils.simple_server_sslcontext()
|
|
with self.assertRaisesRegex(TypeError, 'is not supported'):
|
|
await self.loop.start_tls(None, None, sslctx)
|
|
|
|
self.loop.run_until_complete(main())
|
|
|
|
def test_handshake_timeout(self):
|
|
# bpo-29970: Check that a connection is aborted if handshake is not
|
|
# completed in timeout period, instead of remaining open indefinitely
|
|
client_sslctx = test_utils.simple_client_sslcontext()
|
|
|
|
messages = []
|
|
self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx))
|
|
|
|
server_side_aborted = False
|
|
|
|
def server(sock):
|
|
nonlocal server_side_aborted
|
|
try:
|
|
sock.recv_all(1024 * 1024)
|
|
except ConnectionAbortedError:
|
|
server_side_aborted = True
|
|
finally:
|
|
sock.close()
|
|
|
|
async def client(addr):
|
|
await asyncio.wait_for(
|
|
self.loop.create_connection(
|
|
asyncio.Protocol,
|
|
*addr,
|
|
ssl=client_sslctx,
|
|
server_hostname='',
|
|
ssl_handshake_timeout=support.SHORT_TIMEOUT),
|
|
0.5)
|
|
|
|
with self.tcp_server(server,
|
|
max_clients=1,
|
|
backlog=1) as srv:
|
|
|
|
with self.assertRaises(asyncio.TimeoutError):
|
|
self.loop.run_until_complete(client(srv.addr))
|
|
|
|
self.assertTrue(server_side_aborted)
|
|
|
|
# Python issue #23197: cancelling a handshake must not raise an
|
|
# exception or log an error, even if the handshake failed
|
|
self.assertEqual(messages, [])
|
|
|
|
# The 10s handshake timeout should be cancelled to free related
|
|
# objects without really waiting for 10s
|
|
client_sslctx = weakref.ref(client_sslctx)
|
|
support.gc_collect()
|
|
self.assertIsNone(client_sslctx())
|
|
|
|
def test_create_connection_ssl_slow_handshake(self):
|
|
client_sslctx = test_utils.simple_client_sslcontext()
|
|
|
|
messages = []
|
|
self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx))
|
|
|
|
def server(sock):
|
|
try:
|
|
sock.recv_all(1024 * 1024)
|
|
except ConnectionAbortedError:
|
|
pass
|
|
finally:
|
|
sock.close()
|
|
|
|
async def client(addr):
|
|
reader, writer = await asyncio.open_connection(
|
|
*addr,
|
|
ssl=client_sslctx,
|
|
server_hostname='',
|
|
ssl_handshake_timeout=1.0)
|
|
|
|
with self.tcp_server(server,
|
|
max_clients=1,
|
|
backlog=1) as srv:
|
|
|
|
with self.assertRaisesRegex(
|
|
ConnectionAbortedError,
|
|
r'SSL handshake.*is taking longer'):
|
|
|
|
self.loop.run_until_complete(client(srv.addr))
|
|
|
|
self.assertEqual(messages, [])
|
|
|
|
def test_create_connection_ssl_failed_certificate(self):
|
|
self.loop.set_exception_handler(lambda loop, ctx: None)
|
|
|
|
sslctx = test_utils.simple_server_sslcontext()
|
|
client_sslctx = test_utils.simple_client_sslcontext(
|
|
disable_verify=False)
|
|
|
|
def server(sock):
|
|
try:
|
|
sock.start_tls(
|
|
sslctx,
|
|
server_side=True)
|
|
except ssl.SSLError:
|
|
pass
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
sock.close()
|
|
|
|
async def client(addr):
|
|
reader, writer = await asyncio.open_connection(
|
|
*addr,
|
|
ssl=client_sslctx,
|
|
server_hostname='',
|
|
ssl_handshake_timeout=support.LOOPBACK_TIMEOUT)
|
|
|
|
with self.tcp_server(server,
|
|
max_clients=1,
|
|
backlog=1) as srv:
|
|
|
|
with self.assertRaises(ssl.SSLCertVerificationError):
|
|
self.loop.run_until_complete(client(srv.addr))
|
|
|
|
def test_create_connection_ssl_failed_certificate_no_cycles(self):
|
|
# gh-159049
|
|
sslctx = test_utils.simple_server_sslcontext()
|
|
client_sslctx = test_utils.simple_client_sslcontext(
|
|
disable_verify=False)
|
|
transports = []
|
|
|
|
def server(sock):
|
|
try:
|
|
sock.start_tls(sslctx, server_side=True)
|
|
except OSError:
|
|
pass
|
|
|
|
connection_made = sslproto.SSLProtocol.connection_made
|
|
|
|
def remember_transport(proto, tr):
|
|
transports.append(weakref.ref(tr))
|
|
connection_made(proto, tr)
|
|
|
|
async def conn(addr):
|
|
try:
|
|
await self.loop.create_connection(
|
|
asyncio.Protocol, *addr,
|
|
ssl=client_sslctx, server_hostname='')
|
|
except ssl.SSLCertVerificationError:
|
|
pass
|
|
|
|
support.gc_collect()
|
|
with (self.tcp_server(server) as srv, support.disable_gc(),
|
|
mock.patch.object(sslproto.SSLProtocol, 'connection_made',
|
|
remember_transport)):
|
|
self.loop.run_until_complete(conn(srv.addr))
|
|
self.assertEqual([tr() for tr in transports], [None])
|
|
|
|
def test_start_tls_failed_certificate_no_cycles(self):
|
|
# gh-159049
|
|
sslctx = test_utils.simple_server_sslcontext()
|
|
client_sslctx = test_utils.simple_client_sslcontext(
|
|
disable_verify=False)
|
|
transports = []
|
|
|
|
def server(sock):
|
|
try:
|
|
sock.start_tls(sslctx, server_side=True)
|
|
except OSError:
|
|
pass
|
|
|
|
async def client(addr):
|
|
tr, proto = await self.loop.create_connection(
|
|
asyncio.Protocol, *addr)
|
|
transports.append(weakref.ref(tr))
|
|
try:
|
|
await self.loop.start_tls(
|
|
tr, proto, client_sslctx, server_hostname='')
|
|
except ssl.SSLCertVerificationError:
|
|
pass
|
|
|
|
support.gc_collect()
|
|
with self.tcp_server(server) as srv, support.disable_gc():
|
|
self.loop.run_until_complete(client(srv.addr))
|
|
self.assertEqual([tr() for tr in transports], [None])
|
|
|
|
def test_start_tls_client_corrupted_ssl(self):
|
|
self.loop.set_exception_handler(lambda loop, ctx: None)
|
|
|
|
sslctx = test_utils.simple_server_sslcontext()
|
|
client_sslctx = test_utils.simple_client_sslcontext()
|
|
|
|
def server(sock):
|
|
orig_sock = sock.dup()
|
|
try:
|
|
sock.start_tls(
|
|
sslctx,
|
|
server_side=True)
|
|
sock.sendall(b'A\n')
|
|
sock.recv_all(1)
|
|
orig_sock.send(b'please corrupt the SSL connection')
|
|
except ssl.SSLError:
|
|
pass
|
|
finally:
|
|
orig_sock.close()
|
|
sock.close()
|
|
|
|
async def client(addr):
|
|
reader, writer = await asyncio.open_connection(
|
|
*addr,
|
|
ssl=client_sslctx,
|
|
server_hostname='')
|
|
|
|
self.assertEqual(await reader.readline(), b'A\n')
|
|
writer.write(b'B')
|
|
with self.assertRaises(ssl.SSLError):
|
|
await reader.readline()
|
|
|
|
writer.close()
|
|
return 'OK'
|
|
|
|
with self.tcp_server(server,
|
|
max_clients=1,
|
|
backlog=1) as srv:
|
|
|
|
res = self.loop.run_until_complete(client(srv.addr))
|
|
|
|
self.assertEqual(res, 'OK')
|
|
|
|
|
|
@unittest.skipIf(ssl is None, 'No ssl module')
|
|
class SelectorStartTLSTests(BaseStartTLS, unittest.TestCase):
|
|
|
|
def new_loop(self):
|
|
return asyncio.SelectorEventLoop()
|
|
|
|
|
|
@unittest.skipIf(ssl is None, 'No ssl module')
|
|
@unittest.skipUnless(hasattr(asyncio, 'ProactorEventLoop'), 'Windows only')
|
|
class ProactorStartTLSTests(BaseStartTLS, unittest.TestCase):
|
|
|
|
def new_loop(self):
|
|
return asyncio.ProactorEventLoop()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|