Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
diff --git a/Lib/asyncio/selector_events.py b/Lib/asyncio/selector_events.py
--- a/Lib/asyncio/selector_events.py
+++ b/Lib/asyncio/selector_events.py
@@ -530,14 +530,19 @@ class _SelectorTransport(transports._Flo
try:
self._protocol.connection_lost(exc)
finally:
- self._sock.close()
- self._sock = None
- self._protocol = None
- self._loop = None
- server = self._server
- if server is not None:
- server._detach()
- self._server = None
+ self._finalize_connection()
+
+ def _finalize_connection(self):
+ """Last steps to low-level close and unregister this transport.
+ """
+ self._sock.close()
+ self._sock = None
+ self._protocol = None
+ self._loop = None
+ server = self._server
+ if server is not None:
+ server._detach()
+ self._server = None
def get_write_buffer_size(self):
return len(self._buffer)
@@ -667,7 +672,228 @@ class _SelectorSocketTransport(_Selector
return True
-class _SelectorSslTransport(_SelectorTransport):
+
+class SSLPipe(object):
+ """An SSL "Pipe".
+
+ An SSL pipe allows you to communicate with an SSL/TLS protocol instance
+ through memory buffers. It can be used to implement a security layer for an
+ existing connection where you don't have access to the connection's file
+ descriptor, or for some reason you don't want to use it.
+
+ An SSL pipe can be in "wrapped" and "unwrapped" mode. In unwrapped mode,
+ data is passed through untransformed. In wrapped mode, application level
+ data is encrypted to SSL record level data and vice versa. The SSL record
+ level is the lowest level in the SSL protocol suite and is what travels
+ as-is over the wire.
+
+ An SslPipe initially is in "unwrapped" mode. To start SSL, call
+ :meth:`do_handshake`. To shutdown SSL again, call :meth:`unwrap`.
+ """
+
+ bufsize = 65536
+
+ # This previously used a socketpair to communicate with the SSL protocol
+ # instance but since October 2014 we're using a Memory BIO! This is
+ # cleaner, and more reliable on Windows. See for example issue #12 for more
+ # details.
+
+ S_UNWRAPPED, S_DO_HANDSHAKE, S_WRAPPED, S_SHUTDOWN = range(4)
+
+ def __init__(self, context, server_side, server_hostname=None):
+ """
+ The *context* argument specifies the :class:`ssl.SSLContext` to use.
+ It is recommended to use :func:`~gruvi.ssl.create_ssl_context` so that
+ it will work on all supported Python versions.
+
+ The *server_side* argument indicates whether this is a server side or
+ client side transport.
+
+ The optional *server_hostname* argument can be used to specify the
+ hostname you are connecting to. You may only specify this parameter if
+ the _ssl module supports Server Name Indication (SNI).
+ """
+ self._context = context
+ self._server_side = server_side
+ self._server_hostname = server_hostname
+ self._state = self.S_UNWRAPPED
+ self._incoming = ssl.MemoryBIO()
+ self._outgoing = ssl.MemoryBIO()
+ self._sslobj = None
+ self._need_ssldata = False
+
+ @property
+ def context(self):
+ """The SSL context passed to the constructor."""
+ return self._context
+
+ @property
+ def ssl_object(self):
+ """The internal :class:`ssl.SSLObject` instance."""
+ return self._sslobj
+
+ @property
+ def need_ssldata(self):
+ """Whether more record level data is needed to complete a handshake
+ that is currently in progress."""
+ return self._need_ssldata
+
+ @property
+ def wrapped(self):
+ """Whether a security layer is currently in effect."""
+ return self._state == self.S_WRAPPED
+
+ def do_handshake(self, callback=None):
+ """Start the SSL handshake. Return a list of ssldata.
+
+ The optional *callback* argument can be used to install a callback that
+ will be called when the handshake is complete. The callback will be
+ called with None if successful, else an exception instance.
+ """
+ if self._state != self.S_UNWRAPPED:
+ raise RuntimeError('handshake in progress or completed')
+ wrapargs = ()
+ self._sslobj = self._context.wrap_bio(
+ self._incoming, self._outgoing,
+ server_side=self._server_side,
+ server_hostname=self._server_hostname)
+ self._state = self.S_DO_HANDSHAKE
+ self._on_handshake_complete = callback
+ ssldata, appdata = self.feed_ssldata(b'', only_handshake=True)
+ assert len(appdata) == 0
+ return ssldata
+
+ def shutdown(self, callback=None):
+ """Start the SSL shutdown sequence. Return a list of ssldata.
+
+ The optional *callback* argument can be used to install a callback that
+ will be called when the shutdown is complete. The callback will be
+ called without arguments.
+ """
+ if self._state == self.S_UNWRAPPED:
+ raise RuntimeError('no security layer present')
+ self._state = self.S_SHUTDOWN
+ self._on_handshake_complete = callback
+ ssldata, appdata = self.feed_ssldata(b'')
+ assert appdata == [] or appdata == [b'']
+ return ssldata
+
+ def feed_eof(self):
+ """Send a potentially "ragged" EOF.
+
+ This method will raise an SSL_ERROR_EOF exception if the EOF is
+ unexpected.
+ """
+ self._incoming.write_eof()
+ ssldata, appdata = self.feed_ssldata(b'')
+ assert appdata == [] or appdata == [b'']
+
+ def feed_ssldata(self, data, only_handshake=False):
+ """Feed SSL record level data into the pipe.
+
+ The data must be a bytes instance. It is OK to send an empty bytes
+ instance. This can be used to get ssldata for a handshake initiated by
+ this endpoint.
+
+ Return a (ssldata, appdata) tuple. The ssldata element is a list of
+ buffers containing SSL data that needs to be sent to the remote SSL.
+
+ The appdata element is a list of buffers containing plaintext data that
+ needs to be forwarded to the application. The appdata list may contain
+ an empty buffer indicating an SSL "close_notify" alert. This alert must
+ be acknowledged by calling :meth:`shutdown`.
+ """
+ if self._state == self.S_UNWRAPPED:
+ # If unwrapped, pass plaintext data straight through.
+ return ([], [data] if data else [])
+ ssldata = []; appdata = []
+ self._need_ssldata = False
+ if data:
+ self._incoming.write(data)
+ try:
+ if self._state == self.S_DO_HANDSHAKE:
+ # Call do_handshake() until it doesn't raise anymore.
+ self._sslobj.do_handshake()
+ self._state = self.S_WRAPPED
+ if self._on_handshake_complete:
+ self._on_handshake_complete(None)
+ if only_handshake:
+ return (ssldata, appdata)
+ if self._state == self.S_WRAPPED:
+ # Main state: read data from SSL until close_notify
+ while True:
+ chunk = self._sslobj.read(self.bufsize)
+ appdata.append(chunk)
+ if not chunk: # close_notify
+ break
+ if self._state == self.S_SHUTDOWN:
+ # Call shutdown() until it doesn't raise anymore.
+ self._sslobj.unwrap()
+ self._sslobj = None
+ self._state = self.S_UNWRAPPED
+ if self._on_handshake_complete:
+ self._on_handshake_complete()
+ if self._state == self.S_UNWRAPPED:
+ # Drain possible plaintext data after close_notify.
+ appdata.append(self._incoming.read())
+ except (ssl.SSLError, ssl.CertificateError) as e:
+ if getattr(e, 'errno', -1) not in (
+ ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE,
+ ssl.SSL_ERROR_SYSCALL):
+ if self._state == self.S_DO_HANDSHAKE and self._on_handshake_complete:
+ self._on_handshake_complete(e)
+ raise
+ self._need_ssldata = e.errno == ssl.SSL_ERROR_WANT_READ
+ # Check for record level data that needs to be sent back.
+ # Happens for the initial handshake and renegotiations.
+ if self._outgoing.pending:
+ ssldata.append(self._outgoing.read())
+ return (ssldata, appdata)
+
+ def feed_appdata(self, data, offset=0):
+ """Feed plaintext data into the pipe.
+
+ Return an (ssldata, offset) tuple. The ssldata element is a list of
+ buffers containing record level data that needs to be sent to the
+ remote SSL instance. The offset is the number of plaintext bytes that
+ were processed, which may be less than the length of data.
+
+ NOTE: In case of short writes, this call MUST be retried with the SAME
+ buffer passed into the *data* argument (i.e. the ``id()`` must be the
+ same). This is an OpenSSL requirement. A further particularity is that
+ a short write will always have offset == 0, because the _ssl module
+ does not enable partial writes. And even though the offset is zero,
+ there will still be encrypted data in ssldata.
+ """
+ if self._state == self.S_UNWRAPPED:
+ # pass through data in unwrapped mode
+ return ([data[offset:]] if offset < len(data) else [], len(data))
+ ssldata = []
+ view = memoryview(data)
+ while True:
+ self._need_ssldata = False
+ try:
+ if offset < len(view):
+ offset += self._sslobj.write(view[offset:])
+ except ssl.SSLError as e:
+ # It is not allowed to call write() after unwrap() until the
+ # close_notify is acknowledged. We return the condition to the
+ # caller as a short write.
+ if get_reason(e) == 'PROTOCOL_IS_SHUTDOWN':
+ e.errno = ssl.SSL_ERROR_WANT_READ
+ if e.errno not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE,
+ ssl.SSL_ERROR_SYSCALL):
+ raise
+ self._need_ssldata = e.errno == ssl.SSL_ERROR_WANT_READ
+ # See if there's any record level data back for us.
+ if self._outgoing.pending:
+ ssldata.append(self._outgoing.read())
+ if offset == len(view) or self._need_ssldata:
+ break
+ return (ssldata, offset)
+
+
+class _SelectorSslTransport(_SelectorSocketTransport):
_buffer_factory = bytearray
@@ -697,48 +923,65 @@ class _SelectorSslTransport(_SelectorTra
sslcontext.set_default_verify_paths()
sslcontext.verify_mode = ssl.CERT_REQUIRED
- wrap_kwargs = {
- 'server_side': server_side,
- 'do_handshake_on_connect': False,
- }
- if server_hostname and not server_side and ssl.HAS_SNI:
- wrap_kwargs['server_hostname'] = server_hostname
- sslsock = sslcontext.wrap_socket(rawsock, **wrap_kwargs)
+ if not server_hostname or server_side or not ssl.HAS_SNI:
+ server_hostname = None
- super().__init__(loop, sslsock, protocol, extra, server)
+ _SelectorTransport.__init__(self, loop, rawsock, protocol, extra, server)
+ self._waiter = waiter
+ self._eof = False
+ self._paused = False
+ self._loop.add_reader(self._sock_fd, self._read_ready)
self._server_hostname = server_hostname
- self._waiter = waiter
self._sslcontext = sslcontext
- self._paused = False
-
+ self._sslpipe = SSLPipe(sslcontext, server_side, server_hostname)
+ # App data write buffering (_SelectorSocketTransport handles SSL data).
+ self._write_backlog = []
+ self._write_buffer_size = 0
# SSL-specific extra info. (peercert is set later)
self._extra.update(sslcontext=sslcontext)
+ self._start_handshake()
+
+ def _start_handshake(self):
+ if self._closing:
+ raise TransportError('SSL transport is closing/closed')
if self._loop.get_debug():
logger.debug("%r starts SSL handshake", self)
- start_time = self._loop.time()
+ self._handshake_start_time = self._loop.time()
else:
- start_time = None
- self._on_handshake(start_time)
+ self._handshake_start_time = None
+ self._in_handshake = True
+ self._write_backlog.append([b'', True])
+ self._write_buffer_size += 1
+ self._process_write_backlog()
- def _on_handshake(self, start_time):
+ def _on_handshake_complete(self, handshake_exc):
+ self._in_handshake = False
+ self._loop.remove_reader(self._sock_fd)
+ self._loop.remove_writer(self._sock_fd)
+ sslobj = self._sslpipe.ssl_object
+
+ peercert = None if handshake_exc else sslobj.getpeercert()
+
try:
- self._sock.do_handshake()
- except ssl.SSLWantReadError:
- self._loop.add_reader(self._sock_fd,
- self._on_handshake, start_time)
- return
- except ssl.SSLWantWriteError:
- self._loop.add_writer(self._sock_fd,
- self._on_handshake, start_time)
- return
+ if handshake_exc is not None:
+ raise handshake_exc
+ if not hasattr(self._sslcontext, 'check_hostname'):
+ # Verify hostname if requested, Python 3.4+ uses check_hostname
+ # and checks the hostname in do_handshake()
+ if (self._server_hostname and
+ self._sslcontext.verify_mode != ssl.CERT_NONE):
+ ssl.match_hostname(peercert, self._server_hostname)
except BaseException as exc:
if self._loop.get_debug():
- logger.warning("%r: SSL handshake failed",
- self, exc_info=True)
- self._loop.remove_reader(self._sock_fd)
- self._loop.remove_writer(self._sock_fd)
+ if isinstance(exc, ssl.CertificateError):
+ logger.warning("%r: SSL handshake failed "
+ "on verifying the certificate",
+ self, exc_info=True)
+ else:
+ logger.warning("%r: SSL handshake failed",
+ self, exc_info=True)
self._sock.close()
if self._waiter is not None:
self._waiter.set_exception(exc)
@@ -747,160 +990,140 @@ class _SelectorSslTransport(_SelectorTra
else:
raise
- self._loop.remove_reader(self._sock_fd)
- self._loop.remove_writer(self._sock_fd)
-
- peercert = self._sock.getpeercert()
- if not hasattr(self._sslcontext, 'check_hostname'):
- # Verify hostname if requested, Python 3.4+ uses check_hostname
- # and checks the hostname in do_handshake()
- if (self._server_hostname and
- self._sslcontext.verify_mode != ssl.CERT_NONE):
- try:
- ssl.match_hostname(peercert, self._server_hostname)
- except Exception as exc:
- if self._loop.get_debug():
- logger.warning("%r: SSL handshake failed "
- "on matching the hostname",
- self, exc_info=True)
- self._sock.close()
- if self._waiter is not None:
- self._waiter.set_exception(exc)
- return
-
# Add extra info that becomes available after handshake.
self._extra.update(peercert=peercert,
- cipher=self._sock.cipher(),
- compression=self._sock.compression(),
+ cipher=sslobj.cipher(),
+ compression=sslobj.compression(),
)
- self._read_wants_write = False
- self._write_wants_read = False
+ self._handshake_successful = True
+ self._protocol.connection_made(self)
+ if self._waiter is not None:
+ self._waiter._set_result_unless_cancelled(None)
+
self._loop.add_reader(self._sock_fd, self._read_ready)
- self._loop.call_soon(self._protocol.connection_made, self)
- if self._waiter is not None:
- # wait until protocol.connection_made() has been called
- self._loop.call_soon(self._waiter._set_result_unless_cancelled,
- None)
if self._loop.get_debug():
- dt = self._loop.time() - start_time
+ dt = self._loop.time() - self._handshake_start_time
logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3)
+ def _process_write_backlog(self):
+ # Try to make progress on the write backlog.
+ try:
+ for i in range(len(self._write_backlog)):
+ data, offset = self._write_backlog[0]
+ if data:
+ ssldata, offset = self._sslpipe.feed_appdata(data, offset)
+ elif offset:
+ ssldata, offset = self._sslpipe.do_handshake(self._on_handshake_complete), 1
+ else:
+ ssldata, offset = self._sslpipe.shutdown(self._ssl_eof_received), 1
+ # Temporarily set _closing to False to prevent
+ # _SelectorSocketTransport.write() from raising an error.
+ saved, self._closing = self._closing, False
+ for chunk in ssldata:
+ super().write(chunk)
+ self._closing = saved
+ if offset < len(data):
+ self._write_backlog[0][1] = offset
+ # A short write means that a write is blocked on a read
+ # We need to enable reading if it is not enabled!!
+ assert self._sslpipe.need_ssldata
+ if not self._reading:
+ self.resume_reading()
+ break
+ # An entire chunk from the backlog was processed. We can
+ # delete it and reduce the outstanding buffer size.
+ del self._write_backlog[0]
+ self._write_buffer_size -= offset
+ except BaseException as exc:
+ if self._in_handshake:
+ self._on_handshake_complete(exc)
+ else:
+ self._fatal_error(exc, 'Fatal error on SSL transport')
+
def pause_reading(self):
- # XXX This is a bit icky, given the comment at the top of
- # _read_ready(). Is it possible to evoke a deadlock? I don't
- # know, although it doesn't look like it; write() will still
- # accept more data for the buffer and eventually the app will
- # call resume_reading() again, and things will flow again.
-
if self._closing:
raise RuntimeError('Cannot pause_reading() when closing')
if self._paused:
raise RuntimeError('Already paused')
self._paused = True
- self._loop.remove_reader(self._sock_fd)
+ if not self._sslpipe.need_ssldata:
+ self._loop.remove_reader(self._sock_fd)
if self._loop.get_debug():
logger.debug("%r pauses reading", self)
- def resume_reading(self):
- if not self._paused:
- raise RuntimeError('Not paused')
- self._paused = False
- if self._closing:
- return
- self._loop.add_reader(self._sock_fd, self._read_ready)
- if self._loop.get_debug():
- logger.debug("%r resumes reading", self)
+ def _ssl_eof_received(self):
+ # An SSL EOF is received
+ # XXX we should be able to keep the connection alive (but cleartext)
+ self._eof_received()
+
+ def _eof_received(self):
+ # A lower-level EOF is received
+ try:
+ if not self._in_handshake:
+ if self._loop.get_debug():
+ logger.debug("%r received EOF", self)
+ keep_open = self._protocol.eof_received()
+ if keep_open:
+ logger.warning('returning true from eof_received() '
+ 'has no effect when using ssl')
+ finally:
+ self.close()
+
+ def _call_connection_lost(self, exc):
+ try:
+ if not self._in_handshake:
+ self._protocol.connection_lost(exc)
+ finally:
+ self._finalize_connection()
def _read_ready(self):
- if self._write_wants_read:
- self._write_wants_read = False
- self._write_ready()
-
- if self._buffer:
- self._loop.add_writer(self._sock_fd, self._write_ready)
-
try:
data = self._sock.recv(self.max_size)
- except (BlockingIOError, InterruptedError, ssl.SSLWantReadError):
+ except (BlockingIOError, InterruptedError):
pass
- except ssl.SSLWantWriteError:
- self._read_wants_write = True
- self._loop.remove_reader(self._sock_fd)
- self._loop.add_writer(self._sock_fd, self._write_ready)
except Exception as exc:
self._fatal_error(exc, 'Fatal read error on SSL transport')
else:
if data:
- self._protocol.data_received(data)
+ try:
+ ssldata, appdata = self._sslpipe.feed_ssldata(data)
+ except ssl.SSLError as e:
+ logger.warning('SSL error {} (reason {})', e.errno, e.reason)
+ self.abort()
+ return
+ for chunk in ssldata:
+ super().write(chunk)
+ if appdata and self._paused:
+ self._loop.remove_reader(self._sock_fd)
+ for chunk in appdata:
+ self._protocol.data_received(chunk)
else:
- try:
- if self._loop.get_debug():
- logger.debug("%r received EOF", self)
- keep_open = self._protocol.eof_received()
- if keep_open:
- logger.warning('returning true from eof_received() '
- 'has no effect when using ssl')
- finally:
- self.close()
-
- def _write_ready(self):
- if self._read_wants_write:
- self._read_wants_write = False
- self._read_ready()
-
- if not (self._paused or self._closing):
- self._loop.add_reader(self._sock_fd, self._read_ready)
-
- if self._buffer:
- try:
- n = self._sock.send(self._buffer)
- except (BlockingIOError, InterruptedError, ssl.SSLWantWriteError):
- n = 0
- except ssl.SSLWantReadError:
- n = 0
- self._loop.remove_writer(self._sock_fd)
- self._write_wants_read = True
- except Exception as exc:
- self._loop.remove_writer(self._sock_fd)
- self._buffer.clear()
- self._fatal_error(exc, 'Fatal write error on SSL transport')
- return
-
- if n:
- del self._buffer[:n]
-
- self._maybe_resume_protocol() # May append to buffer.
-
- if not self._buffer:
- self._loop.remove_writer(self._sock_fd)
- if self._closing:
- self._call_connection_lost(None)
+ # End of underlying stream
+ self._eof_received()
def write(self, data):
+ # Write *data* to the transport.
if not isinstance(data, (bytes, bytearray, memoryview)):
- raise TypeError('data argument must be byte-ish (%r)',
- type(data))
+ raise TypeError("data: expecting a bytes-like instance, got {!r}"
+ .format(type(data).__name__))
if not data:
return
+ self._write_backlog.append([data, 0])
+ self._write_buffer_size += len(data)
+ self._maybe_pause_protocol()
+ self._process_write_backlog()
- if self._conn_lost:
- if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
- logger.warning('socket.send() raised exception.')
- self._conn_lost += 1
- return
-
- if not self._buffer:
- self._loop.add_writer(self._sock_fd, self._write_ready)
-
- # Add it to the buffer.
- self._buffer.extend(data)
- self._maybe_pause_protocol()
+ def get_write_buffer_size(self):
+ return self._write_buffer_size
def can_write_eof(self):
return False
+ def write_eof(self):
+ _SelectorTransport.write_eof(self)
+
class _SelectorDatagramTransport(_SelectorTransport):
diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py
--- a/Lib/test/test_asyncio/test_selector_events.py
+++ b/Lib/test/test_asyncio/test_selector_events.py
@@ -1096,17 +1096,16 @@ class SelectorSslTransportTests(test_uti
self.loop = self.new_test_loop()
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
self.sock = mock.Mock(socket.socket)
- self.sock.fileno.return_value = 7
- self.sslsock = mock.Mock()
- self.sslsock.fileno.return_value = 1
+ self.sock.fileno.return_value = 1
+ self.sslobj = mock.Mock()
self.sslcontext = mock.Mock()
- self.sslcontext.wrap_socket.return_value = self.sslsock
+ self.sslcontext.wrap_bio.return_value = self.sslobj
def _make_one(self, create_waiter=None):
transport = _SelectorSslTransport(
self.loop, self.sock, self.protocol, self.sslcontext)
self.sock.reset_mock()
- self.sslsock.reset_mock()
+ self.sslobj.reset_mock()
self.sslcontext.reset_mock()
self.loop.reset_counters()
return transport
@@ -1116,45 +1115,40 @@ class SelectorSslTransportTests(test_uti
tr = _SelectorSslTransport(
self.loop, self.sock, self.protocol, self.sslcontext,
waiter=waiter)
- self.assertTrue(self.sslsock.do_handshake.called)
+ self.assertTrue(self.sslobj.do_handshake.called)
self.loop.assert_reader(1, tr._read_ready)
test_utils.run_briefly(self.loop)
self.assertIsNone(waiter.result())
def test_on_handshake_reader_retry(self):
self.loop.set_debug(False)
- self.sslsock.do_handshake.side_effect = ssl.SSLWantReadError
+ self.sslobj.do_handshake.side_effect = ssl.SSLWantReadError(
+ ssl.SSL_ERROR_WANT_READ, '')
transport = _SelectorSslTransport(
self.loop, self.sock, self.protocol, self.sslcontext)
- self.loop.assert_reader(1, transport._on_handshake, None)
-
- def test_on_handshake_writer_retry(self):
- self.loop.set_debug(False)
- self.sslsock.do_handshake.side_effect = ssl.SSLWantWriteError
- transport = _SelectorSslTransport(
- self.loop, self.sock, self.protocol, self.sslcontext)
- self.loop.assert_writer(1, transport._on_handshake, None)
+ self.loop.assert_reader(1, transport._read_ready)
def test_on_handshake_exc(self):
exc = ValueError()
- self.sslsock.do_handshake.side_effect = exc
+ self.sslobj.do_handshake.side_effect = exc
with test_utils.disable_logger():
waiter = asyncio.Future(loop=self.loop)
transport = _SelectorSslTransport(
self.loop, self.sock, self.protocol, self.sslcontext, waiter)
self.assertTrue(waiter.done())
self.assertIs(exc, waiter.exception())
- self.assertTrue(self.sslsock.close.called)
+ self.assertTrue(self.sock.close.called)
def test_on_handshake_base_exc(self):
transport = _SelectorSslTransport(
self.loop, self.sock, self.protocol, self.sslcontext)
- transport._waiter = asyncio.Future(loop=self.loop)
exc = BaseException()
- self.sslsock.do_handshake.side_effect = exc
+ self.sslobj.do_handshake.side_effect = exc
with test_utils.disable_logger():
- self.assertRaises(BaseException, transport._on_handshake, 0)
- self.assertTrue(self.sslsock.close.called)
+ self.assertRaises(BaseException, transport.__init__,
+ self.loop, self.sock, self.protocol, self.sslcontext,
+ waiter=asyncio.Future(loop=self.loop))
+ self.assertTrue(self.sock.close.called)
self.assertTrue(transport._waiter.done())
self.assertIs(exc, transport._waiter.exception())
@@ -1171,77 +1165,20 @@ class SelectorSslTransportTests(test_uti
with self.assertRaises(RuntimeError):
tr.resume_reading()
- def test_write(self):
- transport = self._make_one()
- transport.write(b'data')
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- def test_write_bytearray(self):
- transport = self._make_one()
- data = bytearray(b'data')
- transport.write(data)
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
- self.assertEqual(data, bytearray(b'data')) # Hasn't been mutated.
- self.assertIsNot(data, transport._buffer) # Hasn't been incorporated.
-
- def test_write_memoryview(self):
- transport = self._make_one()
- data = memoryview(b'data')
- transport.write(data)
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- def test_write_no_data(self):
- transport = self._make_one()
- transport._buffer.extend(b'data')
- transport.write(b'')
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
def test_write_str(self):
transport = self._make_one()
self.assertRaises(TypeError, transport.write, 'str')
- def test_write_closing(self):
+ def test_read_ready_recv(self):
transport = self._make_one()
- transport.close()
- self.assertEqual(transport._conn_lost, 1)
- transport.write(b'data')
- self.assertEqual(transport._conn_lost, 2)
-
- @mock.patch('asyncio.selector_events.logger')
- def test_write_exception(self, m_log):
- transport = self._make_one()
- transport._conn_lost = 1
- transport.write(b'data')
- self.assertEqual(transport._buffer, list_to_buffer())
- transport.write(b'data')
- transport.write(b'data')
- transport.write(b'data')
- transport.write(b'data')
- m_log.warning.assert_called_with('socket.send() raised exception.')
-
- def test_read_ready_recv(self):
- self.sslsock.recv.return_value = b'data'
- transport = self._make_one()
+ transport._sslpipe.feed_ssldata = mock.Mock(
+ return_value=([], [b'data']))
transport._read_ready()
- self.assertTrue(self.sslsock.recv.called)
+ self.assertTrue(self.sock.recv.called)
self.assertEqual((b'data',), self.protocol.data_received.call_args[0])
- def test_read_ready_write_wants_read(self):
- self.loop.add_writer = mock.Mock()
- self.sslsock.recv.side_effect = BlockingIOError
- transport = self._make_one()
- transport._write_wants_read = True
- transport._write_ready = mock.Mock()
- transport._buffer.extend(b'data')
- transport._read_ready()
-
- self.assertFalse(transport._write_wants_read)
- transport._write_ready.assert_called_with()
- self.loop.add_writer.assert_called_with(
- transport._sock_fd, transport._write_ready)
-
def test_read_ready_recv_eof(self):
- self.sslsock.recv.return_value = b''
+ self.sock.recv.return_value = b''
transport = self._make_one()
transport.close = mock.Mock()
transport._read_ready()
@@ -1249,43 +1186,15 @@ class SelectorSslTransportTests(test_uti
self.protocol.eof_received.assert_called_with()
def test_read_ready_recv_conn_reset(self):
- err = self.sslsock.recv.side_effect = ConnectionResetError()
+ err = self.sock.recv.side_effect = ConnectionResetError()
transport = self._make_one()
transport._force_close = mock.Mock()
with test_utils.disable_logger():
transport._read_ready()
transport._force_close.assert_called_with(err)
- def test_read_ready_recv_retry(self):
- self.sslsock.recv.side_effect = ssl.SSLWantReadError
- transport = self._make_one()
- transport._read_ready()
- self.assertTrue(self.sslsock.recv.called)
- self.assertFalse(self.protocol.data_received.called)
-
- self.sslsock.recv.side_effect = BlockingIOError
- transport._read_ready()
- self.assertFalse(self.protocol.data_received.called)
-
- self.sslsock.recv.side_effect = InterruptedError
- transport._read_ready()
- self.assertFalse(self.protocol.data_received.called)
-
- def test_read_ready_recv_write(self):
- self.loop.remove_reader = mock.Mock()
- self.loop.add_writer = mock.Mock()
- self.sslsock.recv.side_effect = ssl.SSLWantWriteError
- transport = self._make_one()
- transport._read_ready()
- self.assertFalse(self.protocol.data_received.called)
- self.assertTrue(transport._read_wants_write)
-
- self.loop.remove_reader.assert_called_with(transport._sock_fd)
- self.loop.add_writer.assert_called_with(
- transport._sock_fd, transport._write_ready)
-
def test_read_ready_recv_exc(self):
- err = self.sslsock.recv.side_effect = OSError()
+ err = self.sock.recv.side_effect = OSError()
transport = self._make_one()
transport._fatal_error = mock.Mock()
transport._read_ready()
@@ -1293,104 +1202,6 @@ class SelectorSslTransportTests(test_uti
err,
'Fatal read error on SSL transport')
- def test_write_ready_send(self):
- self.sslsock.send.return_value = 4
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
- transport._write_ready()
- self.assertEqual(list_to_buffer(), transport._buffer)
- self.assertTrue(self.sslsock.send.called)
-
- def test_write_ready_send_none(self):
- self.sslsock.send.return_value = 0
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data1', b'data2'])
- transport._write_ready()
- self.assertTrue(self.sslsock.send.called)
- self.assertEqual(list_to_buffer([b'data1data2']), transport._buffer)
-
- def test_write_ready_send_partial(self):
- self.sslsock.send.return_value = 2
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data1', b'data2'])
- transport._write_ready()
- self.assertTrue(self.sslsock.send.called)
- self.assertEqual(list_to_buffer([b'ta1data2']), transport._buffer)
-
- def test_write_ready_send_closing_partial(self):
- self.sslsock.send.return_value = 2
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data1', b'data2'])
- transport._write_ready()
- self.assertTrue(self.sslsock.send.called)
- self.assertFalse(self.sslsock.close.called)
-
- def test_write_ready_send_closing(self):
- self.sslsock.send.return_value = 4
- transport = self._make_one()
- transport.close()
- transport._buffer = list_to_buffer([b'data'])
- transport._write_ready()
- self.assertFalse(self.loop.writers)
- self.protocol.connection_lost.assert_called_with(None)
-
- def test_write_ready_send_closing_empty_buffer(self):
- self.sslsock.send.return_value = 4
- transport = self._make_one()
- transport.close()
- transport._buffer = list_to_buffer()
- transport._write_ready()
- self.assertFalse(self.loop.writers)
- self.protocol.connection_lost.assert_called_with(None)
-
- def test_write_ready_send_retry(self):
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
-
- self.sslsock.send.side_effect = ssl.SSLWantWriteError
- transport._write_ready()
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- self.sslsock.send.side_effect = BlockingIOError()
- transport._write_ready()
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- def test_write_ready_send_read(self):
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
-
- self.loop.remove_writer = mock.Mock()
- self.sslsock.send.side_effect = ssl.SSLWantReadError
- transport._write_ready()
- self.assertFalse(self.protocol.data_received.called)
- self.assertTrue(transport._write_wants_read)
- self.loop.remove_writer.assert_called_with(transport._sock_fd)
-
- def test_write_ready_send_exc(self):
- err = self.sslsock.send.side_effect = OSError()
-
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
- transport._fatal_error = mock.Mock()
- transport._write_ready()
- transport._fatal_error.assert_called_with(
- err,
- 'Fatal write error on SSL transport')
- self.assertEqual(list_to_buffer(), transport._buffer)
-
- def test_write_ready_read_wants_write(self):
- self.loop.add_reader = mock.Mock()
- self.sslsock.send.side_effect = BlockingIOError
- transport = self._make_one()
- transport._read_wants_write = True
- transport._read_ready = mock.Mock()
- transport._write_ready()
-
- self.assertFalse(transport._read_wants_write)
- transport._read_ready.assert_called_with()
- self.loop.add_reader.assert_called_with(
- transport._sock_fd, transport._read_ready)
-
def test_write_eof(self):
tr = self._make_one()
self.assertFalse(tr.can_write_eof())
@@ -1413,9 +1224,8 @@ class SelectorSslTransportTests(test_uti
_SelectorSslTransport(
self.loop, self.sock, self.protocol, self.sslcontext,
server_hostname='localhost')
- self.sslcontext.wrap_socket.assert_called_with(
- self.sock, do_handshake_on_connect=False, server_side=False,
- server_hostname='localhost')
+ self.sslcontext.wrap_bio.assert_called_with(
+ mock.ANY, mock.ANY, server_side=False, server_hostname='localhost')
class SelectorSslWithoutSslTransportTests(unittest.TestCase):
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.