.local/lib/python2.7/site-packages/trollius/sslproto.py

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