mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 08:59:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

bpo-29970: Add timeout for SSL handshake in asyncio

Browse source 
10 seconds by default.
This commit is contained in:
Neil Aspinall 2017-12-19 19:45:42 +00:00 • committed by Andrew Svetlov
parent 4b965930e8
commit f7686c1f55
12 changed files with 207 additions and 83 deletions
Download patch file Download diff file Expand all files Collapse all files

14
Lib/test/test_asyncio/test_sslproto.py
View file

@ -11,6 +11,7 @@ except ImportError:
import asyncio
from asyncio import log
from asyncio import sslproto
from asyncio import tasks
from test.test_asyncio import utils as test_utils
@ -25,7 +26,8 @@ class SslProtoHandshakeTests(test_utils.TestCase):
def ssl_protocol(self, waiter=None):
sslcontext = test_utils.dummy_ssl_context()
app_proto = asyncio.Protocol()
proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter)
proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter,
ssl_handshake_timeout=0.1)
self.assertIs(proto._app_transport.get_protocol(), app_proto)
self.addCleanup(proto._app_transport.close)
return proto
@ -63,6 +65,16 @@ class SslProtoHandshakeTests(test_utils.TestCase):
with test_utils.disable_logger():
self.loop.run_until_complete(handshake_fut)
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
ssl_proto = self.ssl_protocol()
transport = self.connection_made(ssl_proto)
with test_utils.disable_logger():
self.loop.run_until_complete(tasks.sleep(0.2, loop=self.loop))
self.assertTrue(transport.abort.called)
def test_eof_received_waiter(self):
waiter = asyncio.Future(loop=self.loop)
ssl_proto = self.ssl_protocol(waiter)