Coverage for /pythoncovmergedfiles/medio/medio/usr/lib/python3.9/asyncio/sslproto.py: 18%
379 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-25 06:05 +0000
1import collections
2import warnings
3try:
4 import ssl
5except ImportError: # pragma: no cover
6 ssl = None
8from . import constants
9from . import protocols
10from . import transports
11from .log import logger
14def _create_transport_context(server_side, server_hostname):
15 if server_side:
16 raise ValueError('Server side SSL needs a valid SSLContext')
18 # Client side may pass ssl=True to use a default
19 # context; in that case the sslcontext passed is None.
20 # The default is secure for client connections.
21 # Python 3.4+: use up-to-date strong settings.
22 sslcontext = ssl.create_default_context()
23 if not server_hostname:
24 sslcontext.check_hostname = False
25 return sslcontext
28# States of an _SSLPipe.
29_UNWRAPPED = "UNWRAPPED"
30_DO_HANDSHAKE = "DO_HANDSHAKE"
31_WRAPPED = "WRAPPED"
32_SHUTDOWN = "SHUTDOWN"
35class _SSLPipe(object):
36 """An SSL "Pipe".
38 An SSL pipe allows you to communicate with an SSL/TLS protocol instance
39 through memory buffers. It can be used to implement a security layer for an
40 existing connection where you don't have access to the connection's file
41 descriptor, or for some reason you don't want to use it.
43 An SSL pipe can be in "wrapped" and "unwrapped" mode. In unwrapped mode,
44 data is passed through untransformed. In wrapped mode, application level
45 data is encrypted to SSL record level data and vice versa. The SSL record
46 level is the lowest level in the SSL protocol suite and is what travels
47 as-is over the wire.
49 An SslPipe initially is in "unwrapped" mode. To start SSL, call
50 do_handshake(). To shutdown SSL again, call unwrap().
51 """
53 max_size = 256 * 1024 # Buffer size passed to read()
55 def __init__(self, context, server_side, server_hostname=None):
56 """
57 The *context* argument specifies the ssl.SSLContext to use.
59 The *server_side* argument indicates whether this is a server side or
60 client side transport.
62 The optional *server_hostname* argument can be used to specify the
63 hostname you are connecting to. You may only specify this parameter if
64 the _ssl module supports Server Name Indication (SNI).
65 """
66 self._context = context
67 self._server_side = server_side
68 self._server_hostname = server_hostname
69 self._state = _UNWRAPPED
70 self._incoming = ssl.MemoryBIO()
71 self._outgoing = ssl.MemoryBIO()
72 self._sslobj = None
73 self._need_ssldata = False
74 self._handshake_cb = None
75 self._shutdown_cb = None
77 @property
78 def context(self):
79 """The SSL context passed to the constructor."""
80 return self._context
82 @property
83 def ssl_object(self):
84 """The internal ssl.SSLObject instance.
86 Return None if the pipe is not wrapped.
87 """
88 return self._sslobj
90 @property
91 def need_ssldata(self):
92 """Whether more record level data is needed to complete a handshake
93 that is currently in progress."""
94 return self._need_ssldata
96 @property
97 def wrapped(self):
98 """
99 Whether a security layer is currently in effect.
101 Return False during handshake.
102 """
103 return self._state == _WRAPPED
105 def do_handshake(self, callback=None):
106 """Start the SSL handshake.
108 Return a list of ssldata. A ssldata element is a list of buffers
110 The optional *callback* argument can be used to install a callback that
111 will be called when the handshake is complete. The callback will be
112 called with None if successful, else an exception instance.
113 """
114 if self._state != _UNWRAPPED:
115 raise RuntimeError('handshake in progress or completed')
116 self._sslobj = self._context.wrap_bio(
117 self._incoming, self._outgoing,
118 server_side=self._server_side,
119 server_hostname=self._server_hostname)
120 self._state = _DO_HANDSHAKE
121 self._handshake_cb = callback
122 ssldata, appdata = self.feed_ssldata(b'', only_handshake=True)
123 assert len(appdata) == 0
124 return ssldata
126 def shutdown(self, callback=None):
127 """Start the SSL shutdown sequence.
129 Return a list of ssldata. A ssldata element is a list of buffers
131 The optional *callback* argument can be used to install a callback that
132 will be called when the shutdown is complete. The callback will be
133 called without arguments.
134 """
135 if self._state == _UNWRAPPED:
136 raise RuntimeError('no security layer present')
137 if self._state == _SHUTDOWN:
138 raise RuntimeError('shutdown in progress')
139 assert self._state in (_WRAPPED, _DO_HANDSHAKE)
140 self._state = _SHUTDOWN
141 self._shutdown_cb = callback
142 ssldata, appdata = self.feed_ssldata(b'')
143 assert appdata == [] or appdata == [b'']
144 return ssldata
146 def feed_eof(self):
147 """Send a potentially "ragged" EOF.
149 This method will raise an SSL_ERROR_EOF exception if the EOF is
150 unexpected.
151 """
152 self._incoming.write_eof()
153 ssldata, appdata = self.feed_ssldata(b'')
154 assert appdata == [] or appdata == [b'']
156 def feed_ssldata(self, data, only_handshake=False):
157 """Feed SSL record level data into the pipe.
159 The data must be a bytes instance. It is OK to send an empty bytes
160 instance. This can be used to get ssldata for a handshake initiated by
161 this endpoint.
163 Return a (ssldata, appdata) tuple. The ssldata element is a list of
164 buffers containing SSL data that needs to be sent to the remote SSL.
166 The appdata element is a list of buffers containing plaintext data that
167 needs to be forwarded to the application. The appdata list may contain
168 an empty buffer indicating an SSL "close_notify" alert. This alert must
169 be acknowledged by calling shutdown().
170 """
171 if self._state == _UNWRAPPED:
172 # If unwrapped, pass plaintext data straight through.
173 if data:
174 appdata = [data]
175 else:
176 appdata = []
177 return ([], appdata)
179 self._need_ssldata = False
180 if data:
181 self._incoming.write(data)
183 ssldata = []
184 appdata = []
185 try:
186 if self._state == _DO_HANDSHAKE:
187 # Call do_handshake() until it doesn't raise anymore.
188 self._sslobj.do_handshake()
189 self._state = _WRAPPED
190 if self._handshake_cb:
191 self._handshake_cb(None)
192 if only_handshake:
193 return (ssldata, appdata)
194 # Handshake done: execute the wrapped block
196 if self._state == _WRAPPED:
197 # Main state: read data from SSL until close_notify
198 while True:
199 chunk = self._sslobj.read(self.max_size)
200 appdata.append(chunk)
201 if not chunk: # close_notify
202 break
204 elif self._state == _SHUTDOWN:
205 # Call shutdown() until it doesn't raise anymore.
206 self._sslobj.unwrap()
207 self._sslobj = None
208 self._state = _UNWRAPPED
209 if self._shutdown_cb:
210 self._shutdown_cb()
212 elif self._state == _UNWRAPPED:
213 # Drain possible plaintext data after close_notify.
214 appdata.append(self._incoming.read())
215 except (ssl.SSLError, ssl.CertificateError) as exc:
216 exc_errno = getattr(exc, 'errno', None)
217 if exc_errno not in (
218 ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE,
219 ssl.SSL_ERROR_SYSCALL):
220 if self._state == _DO_HANDSHAKE and self._handshake_cb:
221 self._handshake_cb(exc)
222 raise
223 self._need_ssldata = (exc_errno == ssl.SSL_ERROR_WANT_READ)
225 # Check for record level data that needs to be sent back.
226 # Happens for the initial handshake and renegotiations.
227 if self._outgoing.pending:
228 ssldata.append(self._outgoing.read())
229 return (ssldata, appdata)
231 def feed_appdata(self, data, offset=0):
232 """Feed plaintext data into the pipe.
234 Return an (ssldata, offset) tuple. The ssldata element is a list of
235 buffers containing record level data that needs to be sent to the
236 remote SSL instance. The offset is the number of plaintext bytes that
237 were processed, which may be less than the length of data.
239 NOTE: In case of short writes, this call MUST be retried with the SAME
240 buffer passed into the *data* argument (i.e. the id() must be the
241 same). This is an OpenSSL requirement. A further particularity is that
242 a short write will always have offset == 0, because the _ssl module
243 does not enable partial writes. And even though the offset is zero,
244 there will still be encrypted data in ssldata.
245 """
246 assert 0 <= offset <= len(data)
247 if self._state == _UNWRAPPED:
248 # pass through data in unwrapped mode
249 if offset < len(data):
250 ssldata = [data[offset:]]
251 else:
252 ssldata = []
253 return (ssldata, len(data))
255 ssldata = []
256 view = memoryview(data)
257 while True:
258 self._need_ssldata = False
259 try:
260 if offset < len(view):
261 offset += self._sslobj.write(view[offset:])
262 except ssl.SSLError as exc:
263 # It is not allowed to call write() after unwrap() until the
264 # close_notify is acknowledged. We return the condition to the
265 # caller as a short write.
266 exc_errno = getattr(exc, 'errno', None)
267 if exc.reason == 'PROTOCOL_IS_SHUTDOWN':
268 exc_errno = exc.errno = ssl.SSL_ERROR_WANT_READ
269 if exc_errno not in (ssl.SSL_ERROR_WANT_READ,
270 ssl.SSL_ERROR_WANT_WRITE,
271 ssl.SSL_ERROR_SYSCALL):
272 raise
273 self._need_ssldata = (exc_errno == ssl.SSL_ERROR_WANT_READ)
275 # See if there's any record level data back for us.
276 if self._outgoing.pending:
277 ssldata.append(self._outgoing.read())
278 if offset == len(view) or self._need_ssldata:
279 break
280 return (ssldata, offset)
283class _SSLProtocolTransport(transports._FlowControlMixin,
284 transports.Transport):
286 _sendfile_compatible = constants._SendfileMode.FALLBACK
288 def __init__(self, loop, ssl_protocol):
289 self._loop = loop
290 # SSLProtocol instance
291 self._ssl_protocol = ssl_protocol
292 self._closed = False
294 def get_extra_info(self, name, default=None):
295 """Get optional transport information."""
296 return self._ssl_protocol._get_extra_info(name, default)
298 def set_protocol(self, protocol):
299 self._ssl_protocol._set_app_protocol(protocol)
301 def get_protocol(self):
302 return self._ssl_protocol._app_protocol
304 def is_closing(self):
305 return self._closed
307 def close(self):
308 """Close the transport.
310 Buffered data will be flushed asynchronously. No more data
311 will be received. After all buffered data is flushed, the
312 protocol's connection_lost() method will (eventually) called
313 with None as its argument.
314 """
315 self._closed = True
316 self._ssl_protocol._start_shutdown()
318 def __del__(self, _warn=warnings.warn):
319 if not self._closed:
320 _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
321 self.close()
323 def is_reading(self):
324 tr = self._ssl_protocol._transport
325 if tr is None:
326 raise RuntimeError('SSL transport has not been initialized yet')
327 return tr.is_reading()
329 def pause_reading(self):
330 """Pause the receiving end.
332 No data will be passed to the protocol's data_received()
333 method until resume_reading() is called.
334 """
335 self._ssl_protocol._transport.pause_reading()
337 def resume_reading(self):
338 """Resume the receiving end.
340 Data received will once again be passed to the protocol's
341 data_received() method.
342 """
343 self._ssl_protocol._transport.resume_reading()
345 def set_write_buffer_limits(self, high=None, low=None):
346 """Set the high- and low-water limits for write flow control.
348 These two values control when to call the protocol's
349 pause_writing() and resume_writing() methods. If specified,
350 the low-water limit must be less than or equal to the
351 high-water limit. Neither value can be negative.
353 The defaults are implementation-specific. If only the
354 high-water limit is given, the low-water limit defaults to an
355 implementation-specific value less than or equal to the
356 high-water limit. Setting high to zero forces low to zero as
357 well, and causes pause_writing() to be called whenever the
358 buffer becomes non-empty. Setting low to zero causes
359 resume_writing() to be called only once the buffer is empty.
360 Use of zero for either limit is generally sub-optimal as it
361 reduces opportunities for doing I/O and computation
362 concurrently.
363 """
364 self._ssl_protocol._transport.set_write_buffer_limits(high, low)
366 def get_write_buffer_size(self):
367 """Return the current size of the write buffer."""
368 return self._ssl_protocol._transport.get_write_buffer_size()
370 @property
371 def _protocol_paused(self):
372 # Required for sendfile fallback pause_writing/resume_writing logic
373 return self._ssl_protocol._transport._protocol_paused
375 def write(self, data):
376 """Write some data bytes to the transport.
378 This does not block; it buffers the data and arranges for it
379 to be sent out asynchronously.
380 """
381 if not isinstance(data, (bytes, bytearray, memoryview)):
382 raise TypeError(f"data: expecting a bytes-like instance, "
383 f"got {type(data).__name__}")
384 if not data:
385 return
386 self._ssl_protocol._write_appdata(data)
388 def can_write_eof(self):
389 """Return True if this transport supports write_eof(), False if not."""
390 return False
392 def abort(self):
393 """Close the transport immediately.
395 Buffered data will be lost. No more data will be received.
396 The protocol's connection_lost() method will (eventually) be
397 called with None as its argument.
398 """
399 self._ssl_protocol._abort()
400 self._closed = True
403class SSLProtocol(protocols.Protocol):
404 """SSL protocol.
406 Implementation of SSL on top of a socket using incoming and outgoing
407 buffers which are ssl.MemoryBIO objects.
408 """
410 def __init__(self, loop, app_protocol, sslcontext, waiter,
411 server_side=False, server_hostname=None,
412 call_connection_made=True,
413 ssl_handshake_timeout=None):
414 if ssl is None:
415 raise RuntimeError('stdlib ssl module not available')
417 if ssl_handshake_timeout is None:
418 ssl_handshake_timeout = constants.SSL_HANDSHAKE_TIMEOUT
419 elif ssl_handshake_timeout <= 0:
420 raise ValueError(
421 f"ssl_handshake_timeout should be a positive number, "
422 f"got {ssl_handshake_timeout}")
424 if not sslcontext:
425 sslcontext = _create_transport_context(
426 server_side, server_hostname)
428 self._server_side = server_side
429 if server_hostname and not server_side:
430 self._server_hostname = server_hostname
431 else:
432 self._server_hostname = None
433 self._sslcontext = sslcontext
434 # SSL-specific extra info. More info are set when the handshake
435 # completes.
436 self._extra = dict(sslcontext=sslcontext)
438 # App data write buffering
439 self._write_backlog = collections.deque()
440 self._write_buffer_size = 0
442 self._waiter = waiter
443 self._loop = loop
444 self._set_app_protocol(app_protocol)
445 self._app_transport = _SSLProtocolTransport(self._loop, self)
446 # _SSLPipe instance (None until the connection is made)
447 self._sslpipe = None
448 self._session_established = False
449 self._in_handshake = False
450 self._in_shutdown = False
451 # transport, ex: SelectorSocketTransport
452 self._transport = None
453 self._call_connection_made = call_connection_made
454 self._ssl_handshake_timeout = ssl_handshake_timeout
456 def _set_app_protocol(self, app_protocol):
457 self._app_protocol = app_protocol
458 self._app_protocol_is_buffer = \
459 isinstance(app_protocol, protocols.BufferedProtocol)
461 def _wakeup_waiter(self, exc=None):
462 if self._waiter is None:
463 return
464 if not self._waiter.cancelled():
465 if exc is not None:
466 self._waiter.set_exception(exc)
467 else:
468 self._waiter.set_result(None)
469 self._waiter = None
471 def connection_made(self, transport):
472 """Called when the low-level connection is made.
474 Start the SSL handshake.
475 """
476 self._transport = transport
477 self._sslpipe = _SSLPipe(self._sslcontext,
478 self._server_side,
479 self._server_hostname)
480 self._start_handshake()
482 def connection_lost(self, exc):
483 """Called when the low-level connection is lost or closed.
485 The argument is an exception object or None (the latter
486 meaning a regular EOF is received or the connection was
487 aborted or closed).
488 """
489 if self._session_established:
490 self._session_established = False
491 self._loop.call_soon(self._app_protocol.connection_lost, exc)
492 else:
493 # Most likely an exception occurred while in SSL handshake.
494 # Just mark the app transport as closed so that its __del__
495 # doesn't complain.
496 if self._app_transport is not None:
497 self._app_transport._closed = True
498 self._transport = None
499 self._app_transport = None
500 if getattr(self, '_handshake_timeout_handle', None):
501 self._handshake_timeout_handle.cancel()
502 self._wakeup_waiter(exc)
503 self._app_protocol = None
504 self._sslpipe = None
506 def pause_writing(self):
507 """Called when the low-level transport's buffer goes over
508 the high-water mark.
509 """
510 self._app_protocol.pause_writing()
512 def resume_writing(self):
513 """Called when the low-level transport's buffer drains below
514 the low-water mark.
515 """
516 self._app_protocol.resume_writing()
518 def data_received(self, data):
519 """Called when some SSL data is received.
521 The argument is a bytes object.
522 """
523 if self._sslpipe is None:
524 # transport closing, sslpipe is destroyed
525 return
527 try:
528 ssldata, appdata = self._sslpipe.feed_ssldata(data)
529 except (SystemExit, KeyboardInterrupt):
530 raise
531 except BaseException as e:
532 self._fatal_error(e, 'SSL error in data received')
533 return
535 for chunk in ssldata:
536 self._transport.write(chunk)
538 for chunk in appdata:
539 if chunk:
540 try:
541 if self._app_protocol_is_buffer:
542 protocols._feed_data_to_buffered_proto(
543 self._app_protocol, chunk)
544 else:
545 self._app_protocol.data_received(chunk)
546 except (SystemExit, KeyboardInterrupt):
547 raise
548 except BaseException as ex:
549 self._fatal_error(
550 ex, 'application protocol failed to receive SSL data')
551 return
552 else:
553 self._start_shutdown()
554 break
556 def eof_received(self):
557 """Called when the other end of the low-level stream
558 is half-closed.
560 If this returns a false value (including None), the transport
561 will close itself. If it returns a true value, closing the
562 transport is up to the protocol.
563 """
564 try:
565 if self._loop.get_debug():
566 logger.debug("%r received EOF", self)
568 self._wakeup_waiter(ConnectionResetError)
570 if not self._in_handshake:
571 keep_open = self._app_protocol.eof_received()
572 if keep_open:
573 logger.warning('returning true from eof_received() '
574 'has no effect when using ssl')
575 finally:
576 self._transport.close()
578 def _get_extra_info(self, name, default=None):
579 if name in self._extra:
580 return self._extra[name]
581 elif self._transport is not None:
582 return self._transport.get_extra_info(name, default)
583 else:
584 return default
586 def _start_shutdown(self):
587 if self._in_shutdown:
588 return
589 if self._in_handshake:
590 self._abort()
591 else:
592 self._in_shutdown = True
593 self._write_appdata(b'')
595 def _write_appdata(self, data):
596 self._write_backlog.append((data, 0))
597 self._write_buffer_size += len(data)
598 self._process_write_backlog()
600 def _start_handshake(self):
601 if self._loop.get_debug():
602 logger.debug("%r starts SSL handshake", self)
603 self._handshake_start_time = self._loop.time()
604 else:
605 self._handshake_start_time = None
606 self._in_handshake = True
607 # (b'', 1) is a special value in _process_write_backlog() to do
608 # the SSL handshake
609 self._write_backlog.append((b'', 1))
610 self._handshake_timeout_handle = \
611 self._loop.call_later(self._ssl_handshake_timeout,
612 self._check_handshake_timeout)
613 self._process_write_backlog()
615 def _check_handshake_timeout(self):
616 if self._in_handshake is True:
617 msg = (
618 f"SSL handshake is taking longer than "
619 f"{self._ssl_handshake_timeout} seconds: "
620 f"aborting the connection"
621 )
622 self._fatal_error(ConnectionAbortedError(msg))
624 def _on_handshake_complete(self, handshake_exc):
625 self._in_handshake = False
626 self._handshake_timeout_handle.cancel()
628 sslobj = self._sslpipe.ssl_object
629 try:
630 if handshake_exc is not None:
631 raise handshake_exc
633 peercert = sslobj.getpeercert()
634 except (SystemExit, KeyboardInterrupt):
635 raise
636 except BaseException as exc:
637 if isinstance(exc, ssl.CertificateError):
638 msg = 'SSL handshake failed on verifying the certificate'
639 else:
640 msg = 'SSL handshake failed'
641 self._fatal_error(exc, msg)
642 return
644 if self._loop.get_debug():
645 dt = self._loop.time() - self._handshake_start_time
646 logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3)
648 # Add extra info that becomes available after handshake.
649 self._extra.update(peercert=peercert,
650 cipher=sslobj.cipher(),
651 compression=sslobj.compression(),
652 ssl_object=sslobj,
653 )
654 if self._call_connection_made:
655 self._app_protocol.connection_made(self._app_transport)
656 self._wakeup_waiter()
657 self._session_established = True
658 # In case transport.write() was already called. Don't call
659 # immediately _process_write_backlog(), but schedule it:
660 # _on_handshake_complete() can be called indirectly from
661 # _process_write_backlog(), and _process_write_backlog() is not
662 # reentrant.
663 self._loop.call_soon(self._process_write_backlog)
665 def _process_write_backlog(self):
666 # Try to make progress on the write backlog.
667 if self._transport is None or self._sslpipe is None:
668 return
670 try:
671 for i in range(len(self._write_backlog)):
672 data, offset = self._write_backlog[0]
673 if data:
674 ssldata, offset = self._sslpipe.feed_appdata(data, offset)
675 elif offset:
676 ssldata = self._sslpipe.do_handshake(
677 self._on_handshake_complete)
678 offset = 1
679 else:
680 ssldata = self._sslpipe.shutdown(self._finalize)
681 offset = 1
683 for chunk in ssldata:
684 self._transport.write(chunk)
686 if offset < len(data):
687 self._write_backlog[0] = (data, offset)
688 # A short write means that a write is blocked on a read
689 # We need to enable reading if it is paused!
690 assert self._sslpipe.need_ssldata
691 if self._transport._paused:
692 self._transport.resume_reading()
693 break
695 # An entire chunk from the backlog was processed. We can
696 # delete it and reduce the outstanding buffer size.
697 del self._write_backlog[0]
698 self._write_buffer_size -= len(data)
699 except (SystemExit, KeyboardInterrupt):
700 raise
701 except BaseException as exc:
702 if self._in_handshake:
703 # Exceptions will be re-raised in _on_handshake_complete.
704 self._on_handshake_complete(exc)
705 else:
706 self._fatal_error(exc, 'Fatal error on SSL transport')
708 def _fatal_error(self, exc, message='Fatal error on transport'):
709 if isinstance(exc, OSError):
710 if self._loop.get_debug():
711 logger.debug("%r: %s", self, message, exc_info=True)
712 else:
713 self._loop.call_exception_handler({
714 'message': message,
715 'exception': exc,
716 'transport': self._transport,
717 'protocol': self,
718 })
719 if self._transport:
720 self._transport._force_close(exc)
722 def _finalize(self):
723 self._sslpipe = None
725 if self._transport is not None:
726 self._transport.close()
728 def _abort(self):
729 try:
730 if self._transport is not None:
731 self._transport.abort()
732 finally:
733 self._finalize()