tests for asynciosslproto py import logging import socket import unittest import weakref from test import support from test support import sockethelper 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 testasyncio import utils as testutils from test testasyncio import functional as functests def teardownmodule asyncio seteventlooppolicynone unittest skipifssl is none no ssl module class sslprotohandshaketeststestutils testcase def setupself super setup self loop asyncio neweventloop self seteventloopself loop def sslprotocolself waiternone protonone sslcontext testutils dummysslcontext if proto is none app protocol proto asyncio protocol sslproto sslproto sslprotocolself loop proto sslcontext waiter sslhandshaketimeout0 1 self assertissslproto apptransport getprotocol proto self addcleanupsslproto apptransport close return sslproto def connectionmadeself sslproto dohandshakenone transport mock mock sslobj mock mock emulate reading decompressed data sslobj read sideeffect ssl sslwantreaderror if dohandshake is not none sslobj dohandshake dohandshake sslproto sslobj sslobj sslproto connectionmadetransport return transport def testhandshaketimeoutzeroself sslcontext testutils dummysslcontext appproto mock mock waiter mock mock with self assertraisesregexvalueerror a positive number sslproto sslprotocolself loop appproto sslcontext waiter sslhandshaketimeout0 def testhandshaketimeoutnegativeself sslcontext testutils dummysslcontext appproto mock mock waiter mock mock with self assertraisesregexvalueerror a positive number sslproto sslprotocolself loop appproto sslcontext waiter sslhandshaketimeout10 def testeofreceivedwaiterself waiter self loop createfuture sslproto self sslprotocolwaiterwaiter self connectionmade sslproto dohandshakemock mocksideeffectssl sslwantreaderror sslproto eofreceived testutils runbrieflyself loop self assertisinstancewaiter exception connectionreseterror def testfatalerrornonameerrorself from issue 363 fatalerror generates a nameerror if sslproto py does not import baseevents waiter self loop createfuture sslproto self sslprotocolwaiterwaiter temporarily turn off error logging so as not to spoil test output loglevel log logger geteffectivelevel log logger setlevellogging fatal try sslproto fatalerrornone finally restore error logging log logger setlevelloglevel def testconnectionlostself from issue 472 yield from waiter hang if lostconnection was called waiter self loop createfuture sslproto self sslprotocolwaiterwaiter self connectionmade sslproto dohandshakemock mocksideeffectssl sslwantreaderror sslproto connectionlostconnectionabortederror testutils runbrieflyself loop self assertisinstancewaiter exception connectionabortederror def testcloseduringhandshakeself bpo29743 closing transport during handshake process leaks socket waiter self loop createfuture sslproto self sslprotocolwaiterwaiter transport self connectionmade sslproto dohandshakemock mocksideeffectssl sslwantreaderror testutils runbrieflyself loop sslproto apptransport close self asserttruetransport abort called def testgetextrainfoonclosedconnectionself waiter self loop createfuture sslproto self sslprotocolwaiterwaiter self assertisnonesslproto getextrainfo socket default object self assertissslproto getextrainfo socket default default self connectionmadesslproto self assertisnotnonesslproto getextrainfo socket sslproto connectionlostnone self assertisnonesslproto getextrainfo socket def testsetnewappprotocolself waiter self loop createfuture sslproto self sslprotocolwaiterwaiter newappproto asyncio protocol sslproto apptransport setprotocolnewappproto self assertissslproto apptransport getprotocol newappproto self assertissslproto appprotocol newappproto def testdatareceivedafterclosingself sslproto self sslprotocol self connectionmadesslproto transp sslproto apptransport transp close should not raise self assertisnonesslproto bufferupdated5 def testwriteafterclosingself sslproto self sslprotocol self connectionmadesslproto transp sslproto apptransport transp close should not raise self assertisnonetransp writeb data start tls tests class basestarttlsfunctests functionaltestcasemixin payloadsize 1024 100 timeout support longtimeout def newloopself raise notimplementederror def testbuffeeddataself class protoasyncio bufferedprotocol def initself bufsize usemv self buf bytearraybufsize self mv memoryviewself buf self data b self usemv usemv def getbufferself sizehint if self usemv return self mv else return self buf def bufferupdatedself nsize if self usemv self data self mv nsize else self data self buf nsize for usemv in false true proto proto1 usemv protocols feeddatatobufferedprotoproto b 12345 self assertequalproto data b 12345 proto proto2 usemv protocols feeddatatobufferedprotoproto b 12345 self assertequalproto data b 12345 proto proto2 usemv protocols feeddatatobufferedprotoproto b 1234 self assertequalproto data b 1234 proto proto4 usemv protocols feeddatatobufferedprotoproto b 1234 self assertequalproto data b 1234 proto proto100 usemv protocols feeddatatobufferedprotoproto b 12345 self assertequalproto data b 12345 proto proto0 usemv with self assertraisesregexruntimeerror empty buffer protocols feeddatatobufferedprotoproto b 12345 def teststarttlsclientregproto1self hellomsg b 1 self payloadsize servercontext testutils simpleserversslcontext clientcontext testutils simpleclientsslcontext def servesock sock settimeoutself timeout data sock recvalllenhellomsg self assertequallendata lenhellomsg sock starttlsservercontext serversidetrue sock sendallb o data sock recvalllenhellomsg self assertequallendata lenhellomsg sock shutdownsocket shutrdwr sock close class clientprotoasyncio protocol def initself ondata oneof self ondata ondata self oneof oneof self conmadecnt 0 def connectionmadeproto tr proto conmadecnt 1 ensure connectionmade gets called only once self assertequalproto conmadecnt 1 def datareceivedself data self ondata setresultdata def eofreceivedself self oneof setresulttrue async def clientaddr await asyncio sleep0 5 ondata self loop createfuture oneof self loop createfuture tr proto await self loop createconnection lambda clientprotoondata oneof addr tr writehellomsg newtr await self loop starttlstr proto clientcontext self assertequalawait ondata b o newtr writehellomsg await oneof newtr close with self tcpserverserve timeoutself timeout as srv self loop rununtilcomplete asyncio waitforclientsrv addr timeoutsupport shorttimeout no garbage is left if ssl is closed uncleanly clientcontext weakref refclientcontext support gccollect self assertisnoneclientcontext def testcreateconnectionmemoryleakself hellomsg b 1 self payloadsize servercontext testutils simpleserversslcontext clientcontext testutils simpleclientsslcontext def servesock sock settimeoutself timeout sock starttlsservercontext serversidetrue sock sendallb o data sock recvalllenhellomsg self assertequallendata lenhellomsg sock shutdownsocket shutrdwr sock close class clientprotoasyncio protocol def initself ondata oneof self ondata ondata self oneof oneof self conmadecnt 0 def connectionmadeproto tr xxx we assume user stores the transport in protocol proto tr tr proto conmadecnt 1 ensure connectionmade gets called only once self assertequalproto conmadecnt 1 def datareceivedself data self ondata setresultdata def eofreceivedself self oneof setresulttrue async def clientaddr await asyncio sleep0 5 ondata self loop createfuture oneof self loop createfuture tr proto await self loop createconnection lambda clientprotoondata oneof addr sslclientcontext self assertequalawait ondata b o tr writehellomsg await oneof tr close with self tcpserverserve timeoutself timeout as srv self loop rununtilcomplete asyncio waitforclientsrv addr timeoutsupport shorttimeout no garbage is left for ssl client from loop createconnection even if user stores the ssltransport in corresponding protocol instance clientcontext weakref refclientcontext support gccollect self assertisnoneclientcontext sockethelper skipiftcpblackhole def teststarttlsclientbufproto1self hellomsg b 1 self payloadsize servercontext testutils simpleserversslcontext clientcontext testutils simpleclientsslcontext clientconmadecalls 0 def servesock sock settimeoutself timeout data sock recvalllenhellomsg self assertequallendata lenhellomsg sock starttlsservercontext serversidetrue sock sendallb o data sock recvalllenhellomsg self assertequallendata lenhellomsg sock sendallb 2 data sock recvalllenhellomsg self assertequallendata lenhellomsg sock shutdownsocket shutrdwr sock close class clientprotofirstasyncio bufferedprotocol def initself ondata self ondata ondata self buf bytearray1 def connectionmadeself tr nonlocal clientconmadecalls clientconmadecalls 1 def getbufferself sizehint return self buf def bufferupdatedslf nsize self assertequalnsize 1 slf ondata setresultbytesslf buf nsize class clientprotosecondasyncio protocol def initself ondata oneof self ondata ondata self oneof oneof self conmadecnt 0 def connectionmadeself tr nonlocal clientconmadecalls clientconmadecalls 1 def datareceivedself data self ondata setresultdata def eofreceivedself self oneof setresulttrue async def clientaddr await asyncio sleep0 5 ondata1 self loop createfuture ondata2 self loop createfuture oneof self loop createfuture tr proto await self loop createconnection lambda clientprotofirstondata1 addr tr writehellomsg newtr await self loop starttlstr proto clientcontext self assertequalawait ondata1 b o newtr writehellomsg newtr setprotocolclientprotosecondondata2 oneof self assertequalawait ondata2 b 2 newtr writehellomsg await oneof newtr close connectionmade should be called only once when we establish connection for the first time start tls doesn t call connectionmade on application protocols self assertequalclientconmadecalls 1 with self tcpserverserve timeoutself timeout as srv self loop rununtilcomplete asyncio waitforclientsrv addr timeoutself timeout def teststarttlsslowclientcancelself hellomsg b 1 self payloadsize clientcontext testutils simpleclientsslcontext serverwaitsonhandshake self loop createfuture def servesock sock settimeoutself timeout data sock recvalllenhellomsg self assertequallendata lenhellomsg try self loop callsoonthreadsafe serverwaitsonhandshake setresult none data sock recvall1024 1024 except connectionabortederror pass finally sock close class clientprotoasyncio protocol def initself ondata oneof self ondata ondata self oneof oneof self conmadecnt 0 def connectionmadeproto tr proto conmadecnt 1 ensure connectionmade gets called only once self assertequalproto conmadecnt 1 def datareceivedself data self ondata setresultdata def eofreceivedself self oneof setresulttrue async def clientaddr await asyncio sleep0 5 ondata self loop createfuture oneof self loop createfuture tr proto await self loop createconnection lambda clientprotoondata oneof addr tr writehellomsg await serverwaitsonhandshake with self assertraisesasyncio timeouterror await asyncio waitfor self loop starttlstr proto clientcontext 0 5 with self tcpserverserve timeoutself timeout as srv self loop rununtilcomplete asyncio waitforclientsrv addr timeoutsupport shorttimeout sockethelper skipiftcpblackhole def teststarttlsserver1self hellomsg b 1 self payloadsize answer b answer servercontext testutils simpleserversslcontext clientcontext testutils simpleclientsslcontext answer none def clientsock addr nonlocal answer sock settimeoutself timeout sock connectaddr data sock recvalllenhellomsg self assertequallendata lenhellomsg sock starttlsclientcontext sock sendallhellomsg answer sock recvalllenanswer sock close class serverprotoasyncio protocol def initself oncon onconlost ongothello self oncon oncon self onconlost onconlost self ongothello ongothello self data b self transport none def connectionmadeself tr self transport tr self oncon setresulttr def replacetransportself tr self transport tr def datareceivedself data self data data if lenself data lenhellomsg self ongothello setresultnone def connectionlostself exc self transport none if exc is none self onconlost setresultnone else self onconlost setexceptionexc async def mainproto oncon onconlost ongothello tr await oncon tr writehellomsg self assertequalproto data b newtr await self loop starttls tr proto servercontext serversidetrue sslhandshaketimeoutself timeout proto replacetransportnewtr await ongothello newtr writeanswer await onconlost self assertequalproto data hellomsg newtr close async def runmain oncon self loop createfuture onconlost self loop createfuture ongothello self loop createfuture proto serverprotooncon onconlost ongothello server await self loop createserver lambda proto 127 0 0 1 0 addr server sockets0 getsockname with self tcpclientlambda sock clientsock addr timeoutself timeout await asyncio waitfor mainproto oncon onconlost ongothello timeoutself timeout server close await server waitclosed self assertequalanswer answer self loop rununtilcompleterunmain def teststarttlswrongargsself async def main with self assertraisesregextypeerror sslcontext got await self loop starttlsnone none none sslctx testutils simpleserversslcontext with self assertraisesregextypeerror is not supported await self loop starttlsnone none sslctx self loop rununtilcompletemain def testhandshaketimeoutself bpo29970 check that a connection is aborted if handshake is not completed in timeout period instead of remaining open indefinitely clientsslctx testutils simpleclientsslcontext messages self loop setexceptionhandlerlambda loop ctx messages appendctx serversideaborted false def serversock nonlocal serversideaborted try sock recvall1024 1024 except connectionabortederror serversideaborted true finally sock close async def clientaddr await asyncio waitfor self loop createconnection asyncio protocol addr sslclientsslctx serverhostname sslhandshaketimeoutsupport shorttimeout 0 5 with self tcpserverserver maxclients1 backlog1 as srv with self assertraisesasyncio timeouterror self loop rununtilcompleteclientsrv addr self asserttrueserversideaborted python issue 23197 cancelling a handshake must not raise an exception or log an error even if the handshake failed self assertequalmessages the 10s handshake timeout should be cancelled to free related objects without really waiting for 10s clientsslctx weakref refclientsslctx support gccollect self assertisnoneclientsslctx def testcreateconnectionsslslowhandshakeself clientsslctx testutils simpleclientsslcontext messages self loop setexceptionhandlerlambda loop ctx messages appendctx def serversock try sock recvall1024 1024 except connectionabortederror pass finally sock close async def clientaddr reader writer await asyncio openconnection addr sslclientsslctx serverhostname sslhandshaketimeout1 0 with self tcpserverserver maxclients1 backlog1 as srv with self assertraisesregex connectionabortederror r ssl handshake is taking longer self loop rununtilcompleteclientsrv addr self assertequalmessages def testcreateconnectionsslfailedcertificateself self loop setexceptionhandlerlambda loop ctx none sslctx testutils simpleserversslcontext clientsslctx testutils simpleclientsslcontext disableverifyfalse def serversock try sock starttls sslctx serversidetrue except ssl sslerror pass except oserror pass finally sock close async def clientaddr reader writer await asyncio openconnection addr sslclientsslctx serverhostname sslhandshaketimeoutsupport loopbacktimeout with self tcpserverserver maxclients1 backlog1 as srv with self assertraisesssl sslcertverificationerror self loop rununtilcompleteclientsrv addr def teststarttlsclientcorruptedsslself self loop setexceptionhandlerlambda loop ctx none sslctx testutils simpleserversslcontext clientsslctx testutils simpleclientsslcontext def serversock origsock sock dup try sock starttls sslctx serversidetrue sock sendallb an sock recvall1 origsock sendb please corrupt the ssl connection except ssl sslerror pass finally origsock close sock close async def clientaddr reader writer await asyncio openconnection addr sslclientsslctx serverhostname self assertequalawait reader readline b an writer writeb b with self assertraisesssl sslerror await reader readline writer close return ok with self tcpserverserver maxclients1 backlog1 as srv res self loop rununtilcompleteclientsrv addr self assertequalres ok unittest skipifssl is none no ssl module class selectorstarttlstestsbasestarttls unittest testcase def newloopself return asyncio selectoreventloop unittest skipifssl is none no ssl module unittest skipunlesshasattrasyncio proactoreventloop windows only class proactorstarttlstestsbasestarttls unittest testcase def newloopself return asyncio proactoreventloop if name main unittest main app protocol emulate reading decompressed data from issue 363 _fatal_error generates a nameerror if sslproto py does not import base_events temporarily turn off error logging so as not to spoil test output restore error logging from issue 472 yield from waiter hang if lost_connection was called bpo 29743 closing transport during handshake process leaks socket should not raise should not raise start tls tests ensure connection_made gets called only once no garbage is left if ssl is closed uncleanly xxx we assume user stores the transport in protocol ensure connection_made gets called only once no garbage is left for ssl client from loop create_connection even if user stores the ssltransport in corresponding protocol instance 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 ensure connection_made gets called only once bpo 29970 check that a connection is aborted if handshake is not completed in timeout period instead of remaining open indefinitely python issue 23197 cancelling a handshake must not raise an exception or log an error even if the handshake failed the 10s handshake timeout should be cancelled to free related objects without really waiting for 10s
import logging
import socket
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.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:  
            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()
        
        sslobj.read.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_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):
        
        
        
        waiter = self.loop.create_future()
        ssl_proto = self.ssl_protocol(waiter=waiter)
        
        log_level = log.logger.getEffectiveLevel()
        log.logger.setLevel(logging.FATAL)
        try:
            ssl_proto._fatal_error(None)
        finally:
            
            log.logger.setLevel(log_level)

    def test_connection_lost(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.connection_lost(ConnectionAbortedError)
        test_utils.run_briefly(self.loop)
        self.assertIsInstance(waiter.exception(), ConnectionAbortedError)

    def test_close_during_handshake(self):
        
        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.abort.called)

    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()

        
        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()

        
        self.assertIsNone(transp.write(b'data'))







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
                
                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))

        
        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):
                
                proto.tr = tr
                proto.con_made_cnt += 1
                
                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))

        
        
        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()

            
            
            
            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):
                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
                
                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)

            await server_waits_on_handshake

            with self.assertRaises(asyncio.TimeoutError):
                await asyncio.wait_for(
                    self.loop.start_tls(tr, proto, client_context),
                    0.5)

        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))

    @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):
        
        
        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)

        
        
        self.assertEqual(messages, [])

        
        
        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_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()