sslproto.py
python
import collections
import warnings
try:
import ssl
except ImportError: # pragma: no cover
ssl = None
from . import base_events
from . import compat
from . import protocols
from . import transports
from .log import logger
def _create_transport_context(server_side, server_hostname):
if server_side:
raise ValueError('Server side SSL needs a valid SSLContext')
# Клиентская сторона может передать ssl=True, чтобы использовать умолчательный
# контекст; в этом случае переданный sslcontext равен None.
# По умолчанию используется безопасный контекст для клиентских соединений.
if hasattr(ssl, 'create_default_context'):
# Python 3.4+: использовать современные надежные настройки.
sslcontext = ssl.create_default_context()
if not server_hostname:
sslcontext.check_hostname = False
else:
# Запасной вариант для Python 3.3.
sslcontext = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
sslcontext.options |= ssl.OP_NO_SSLv2
sslcontext.options |= ssl.OP_NO_SSLv3
sslcontext.set_default_verify_paths()
sslcontext.verify_mode = ssl.CERT_REQUIRED
return sslcontext
def _is_sslproto_available():
return hasattr(ssl, "MemoryBIO")
# Состояния объекта _SSLPipe.
_UNWRAPPED = "UNWRAPPED"
_DO_HANDSHAKE = "DO_HANDSHAKE"
_WRAPPED = "WRAPPED"
_SHUTDOWN = "SHUTDOWN"
class _SSLPipe(object):
"""SSL-«Канал».
SSL-канал позволяет общаться с экземпляром протокола SSL/TLS через буферы в памяти. Его можно использовать для реализации уровня безопасности для существующего соединения, когда у вас нет доступа к файловому дескриптору соединения или по какой-то причине вы не хотите его использовать.
SSL-канал может находиться в режиме «обёрнутого» (wrapped) и «развёрнутого» (unwrapped). В развёрнутом режиме данные передаются без преобразования. В обёрнутом режиме данные прикладного уровня шифруются до уровня записи SSL и наоборот. Уровень SSL-записей — это самый нижний уровень в наборе протоколов SSL, и именно он передаётся по сети без изменений.
Изначально SslPipe находится в «развёрнутом» режиме. Для запуска SSL вызовите do_handshake(). Чтобы снова завершить SSL, вызовите unwrap().
"""
max_size = 256 * 1024 # Размер буфера, передаваемый в read()
def __init__(self, context, server_side, server_hostname=None):
"""
Аргумент *context* указывает используемый ssl.SSLContext.
Аргумент *server_side* указывает, является ли транспорт серверным или клиентским.
Необязательный аргумент *server_hostname* может использоваться для указания имени хоста, к которому вы подключаетесь. Этот параметр можно указывать только в том случае, если модуль _ssl поддерживает Server Name Indication (SNI).
"""
self._context = context
self._server_side = server_side
self._server_hostname = server_hostname
self._state = _UNWRAPPED
self._incoming = ssl.MemoryBIO()
self._outgoing = ssl.MemoryBIO()
self._sslobj = None
self._need_ssldata = False
self._handshake_cb = None
self._shutdown_cb = None
@property
def context(self):
"""Контекст SSL, переданный конструктору."""
return self._context
@property
def ssl_object(self):
"""Внутренний экземпляр ssl.SSLObject.
Возвращает None, если канал не обёрнут.
"""
return self._sslobj
@property
def need_ssldata(self):
"""Нужно ли больше данных уровня записей для завершения рукопожатия, выполняющегося в данный момент."""
return self._need_ssldata
@property
def wrapped(self):
"""
Действует ли в данный момент уровень безопасности. Возвращает False во время рукопожатия.
"""
return self._state == _WRAPPED
def do_handshake(self, callback=None):
"""Начать SSL-рукопожатие.
Возвращает список ssldata. Элемент ssldata — это список буферов.
Необязательный аргумент *callback* можно использовать для установки колбэка, который будет вызван при завершении рукопожатия. Колбэк будет вызван с None в случае успеха, иначе с экземпляром исключения.
"""
if self._state != _UNWRAPPED:
raise RuntimeError('handshake in progress or completed')
self._sslobj = self._context.wrap_bio(
self._incoming, self._outgoing,
server_side=self._server_side,
server_hostname=self._server_hostname)
self._state = _DO_HANDSHAKE
self._handshake_cb = callback
ssldata, appdata = self.feed_ssldata(b'', only_handshake=True)
assert len(appdata) == 0
return ssldata
def shutdown(self, callback=None):
"""Начать последовательность завершения SSL.
Возвращает список ssldata. Элемент ssldata — это список буферов.
Необязательный аргумент *callback* можно использовать для установки колбэка, который будет вызван при завершении завершения. Колбэк будет вызван без аргументов.
"""
if self._state == _UNWRAPPED:
raise RuntimeError('no security layer present')
if self._state == _SHUTDOWN:
raise RuntimeError('shutdown in progress')
assert self._state in (_WRAPPED, _DO_HANDSHAKE)
self._state = _SHUTDOWN
self._shutdown_cb = callback
ssldata, appdata = self.feed_ssldata(b'')
assert appdata == [] or appdata == [b'']
return ssldata
def feed_eof(self):
"""Отправить потенциально «рваный» EOF.
Этот метод вызовет исключение SSL_ERROR_EOF, если EOF неожиданный.
"""
self._incoming.write_eof()
ssldata, appdata = self.feed_ssldata(b'')
assert appdata == [] or appdata == [b'']
def feed_ssldata(self, data, only_handshake=False):
"""Передать данные уровня записей SSL в канал.
Данные должны быть экземпляром bytes. Допустимо отправлять пустой экземпляр bytes. Это можно использовать для получения ssldata для рукопожатия, инициированного этой конечной точкой.
Возвращает кортеж (ssldata, appdata). Элемент ssldata — это список буферов, содержащих SSL-данные, которые необходимо отправить удалённому SSL.
Элемент appdata — это список буферов, содержащих незашифрованные данные, которые необходимо передать приложению. Список appdata может содержать пустой буфер, указывающий на предупреждение SSL «close_notify». Это предупреждение должно быть подтверждено вызовом shutdown().
"""
if self._state == _UNWRAPPED:
# Если оболочка снята, передавать незашифрованные данные напрямую.
if data:
appdata = [data]
else:
appdata = []
return ([], appdata)
self._need_ssldata = False
if data:
self._incoming.write(data)
ssldata = []
appdata = []
try:
if self._state == _DO_HANDSHAKE:
# Вызывать do_handshake(), пока он не перестанет вызывать исключения.
self._sslobj.do_handshake()
self._state = _WRAPPED
if self._handshake_cb:
self._handshake_cb(None)
if only_handshake:
return (ssldata, appdata)
# Рукопожатие завершено: выполнить обёрнутый блок
if self._state == _WRAPPED:
# Основное состояние: читать данные из SSL до close_notify
while True:
chunk = self._sslobj.read(self.max_size)
appdata.append(chunk)
if not chunk: # close_notify
break
elif self._state == _SHUTDOWN:
# Вызывать shutdown(), пока он не перестанет вызывать исключения.
self._sslobj.unwrap()
self._sslobj = None
self._state = _UNWRAPPED
if self._shutdown_cb:
self._shutdown_cb()
elif self._state == _UNWRAPPED:
# Слить возможные незашифрованные данные после close_notify.
appdata.append(self._incoming.read())
except (ssl.SSLError, ssl.CertificateError) as exc:
if getattr(exc, 'errno', None) not in (
ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE,
ssl.SSL_ERROR_SYSCALL):
if self._state == _DO_HANDSHAKE and self._handshake_cb:
self._handshake_cb(exc)
raise
self._need_ssldata = (exc.errno == ssl.SSL_ERROR_WANT_READ)
# Проверить данные уровня записей, которые нужно отправить обратно.
# Происходит для начального рукопожатия и пересогласований.
if self._outgoing.pending:
ssldata.append(self._outgoing.read())
return (ssldata, appdata)
def feed_appdata(self, data, offset=0):
"""Передать незашифрованные данные в канал.
Возвращает кортеж (ssldata, offset). Элемент ssldata — это список буферов, содержащих данные уровня записей, которые необходимо отправить удалённому экземпляру SSL. offset — это количество обработанных байтов незашифрованных данных, которое может быть меньше длины данных.
ПРИМЕЧАНИЕ: В случае короткой записи этот вызов ОБЯЗАТЕЛЬНО нужно повторить с тем же буфером, переданным в аргумент *data* (т.е. id() должен быть тем же). Это требование OpenSSL. Ещё одна особенность: при короткой записи offset всегда равен 0, потому что модуль _ssl не поддерживает частичные записи. И даже если offset равен нулю, в ssldata всё равно будут зашифрованные данные.
"""
assert 0 <= offset <= len(data)
if self._state == _UNWRAPPED:
# передавать данные в режиме без обёртки
if offset < len(data):
ssldata = [data[offset:]]
else:
ssldata = []
return (ssldata, 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 exc:
# Не разрешается вызывать write() после unwrap() до тех пор, пока close_notify не будет подтверждён. Мы возвращаем это условие вызывающей стороне как короткую запись.
# close_notify подтверждён. Мы возвращаем условие
# вызывающей стороне как короткая запись.
if exc.reason == 'PROTOCOL_IS_SHUTDOWN':
exc.errno = ssl.SSL_ERROR_WANT_READ
if exc.errno not in (ssl.SSL_ERROR_WANT_READ,
ssl.SSL_ERROR_WANT_WRITE,
ssl.SSL_ERROR_SYSCALL):
raise
self._need_ssldata = (exc.errno == ssl.SSL_ERROR_WANT_READ)
# Проверить, есть ли какие-либо данные уровня записи для нас.
if self._outgoing.pending:
ssldata.append(self._outgoing.read())
if offset == len(view) or self._need_ssldata:
break
return (ssldata, offset)
class _SSLProtocolTransport(transports._FlowControlMixin,
transports.Transport):
def __init__(self, loop, ssl_protocol):
self._loop = loop
# экземпляр SSLProtocol
self._ssl_protocol = ssl_protocol
self._closed = False
def get_extra_info(self, name, default=None):
"""Получить дополнительную информацию о транспорте."""
return self._ssl_protocol._get_extra_info(name, default)
def set_protocol(self, protocol):
self._ssl_protocol._app_protocol = protocol
def get_protocol(self):
return self._ssl_protocol._app_protocol
def is_closing(self):
return self._closed
def close(self):
"""Закрыть транспорт.
Буферизованные данные будут асинхронно сброшены. Больше данных
не будет получено. После того, как все буферизованные данные будут сброшены,
метод connection_lost() протокола будет (в конечном итоге) вызван
с None в качестве аргумента.
"""
self._closed = True
self._ssl_protocol._start_shutdown()
# В Python 3.3 и старше объекты с деструктором, являющиеся частью циклической ссылки,
# никогда не уничтожаются. В Python 3.4 это уже не так благодаря
# PEP 442.
if compat.PY34:
def __del__(self):
if not self._closed:
warnings.warn("unclosed transport %r" % self, ResourceWarning,
source=self)
self.close()
def pause_reading(self):
"""Приостановить приём.
Данные не будут передаваться методу data_received() протокола
до тех пор, пока не будет вызван resume_reading().
"""
self._ssl_protocol._transport.pause_reading()
def resume_reading(self):
"""Возобновить приём.
Полученные данные снова будут передаваться методу data_received()
протокола.
"""
self._ssl_protocol._transport.resume_reading()
def set_write_buffer_limits(self, high=None, low=None):
"""Установить верхний и нижний пороги для управления потоком записи.
Эти два значения определяют, когда вызывать методы pause_writing()
и resume_writing() протокола. Если они заданы,
нижний порог должен быть меньше или равен
верхнему порогу. Ни одно из значений не может быть отрицательным.
Значения по умолчанию зависят от реализации. Если указан только
верхний порог, нижний порог по умолчанию устанавливается в
значение, зависящее от реализации, меньшее или равное
верхнему порогу. Установка верхнего порога в ноль также обнуляет
нижний порог и приводит к вызову pause_writing() каждый раз,
когда буфер становится непустым. Установка нижнего порога в ноль
приводит к вызову resume_writing() только после опустошения буфера.
Использование нуля для любого порога, как правило, неоптимально, так как
это уменьшает возможности для одновременного выполнения операций ввода-вывода
и вычислений.
"""
self._ssl_protocol._transport.set_write_buffer_limits(high, low)
def get_write_buffer_size(self):
"""Вернуть текущий размер буфера записи."""
return self._ssl_protocol._transport.get_write_buffer_size()
def write(self, data):
"""Записать некоторые байты данных в транспорт.
Этот вызов не блокируется; он буферизует данные и обеспечивает их
асинхронную отправку.
"""
if not isinstance(data, (bytes, bytearray, memoryview)):
raise TypeError("data: expecting a bytes-like instance, got {!r}"
.format(type(data).__name__))
if not data:
return
self._ssl_protocol._write_appdata(data)
def can_write_eof(self):
"""Вернуть True, если этот транспорт поддерживает write_eof(), иначе False."""
return False
def abort(self):
"""Закрыть транспорт немедленно.
Буферизованные данные будут потеряны. Новые данные приниматься не будут.
Метод connection_lost() протокола будет (в конечном итоге) вызван
с None в качестве аргумента.
"""
self._ssl_protocol._abort()
class SSLProtocol(protocols.Protocol):
"""SSL-протокол.
Реализация SSL поверх сокета с использованием входящих и исходящих
буферов, которые являются объектами ssl.MemoryBIO.
"""
def __init__(self, loop, app_protocol, sslcontext, waiter,
server_side=False, server_hostname=None,
call_connection_made=True):
if ssl is None:
raise RuntimeError('stdlib ssl module not available')
if not sslcontext:
sslcontext = _create_transport_context(server_side, server_hostname)
self._server_side = server_side
if server_hostname and not server_side:
self._server_hostname = server_hostname
else:
self._server_hostname = None
self._sslcontext = sslcontext
# Дополнительная информация, специфичная для SSL. Дополнительные сведения устанавливаются при рукопожатии
# завершается.
self._extra = dict(sslcontext=sslcontext)
# Буферизация записи данных приложения
self._write_backlog = collections.deque()
self._write_buffer_size = 0
self._waiter = waiter
self._loop = loop
self._app_protocol = app_protocol
self._app_transport = _SSLProtocolTransport(self._loop, self)
# _SSLPipe экземпляр (None до установления соединения)
self._sslpipe = None
self._session_established = False
self._in_handshake = False
self._in_shutdown = False
# транспорт, например: SelectorSocketTransport
self._transport = None
self._call_connection_made = call_connection_made
def _wakeup_waiter(self, exc=None):
if self._waiter is None:
return
if not self._waiter.cancelled():
if exc is not None:
self._waiter.set_exception(exc)
else:
self._waiter.set_result(None)
self._waiter = None
def connection_made(self, transport):
"""Вызывается при установлении низкоуровневого соединения.
Запустить SSL-рукопожатие.
"""
self._transport = transport
self._sslpipe = _SSLPipe(self._sslcontext,
self._server_side,
self._server_hostname)
self._start_handshake()
def connection_lost(self, exc):
"""Вызывается при потере или закрытии низкоуровневого соединения.
Аргументом является объект исключения или None (последнее
означает получение обычного EOF или что соединение было
прервано или закрыто).
"""
if self._session_established:
self._session_established = False
self._loop.call_soon(self._app_protocol.connection_lost, exc)
self._transport = None
self._app_transport = None
self._wakeup_waiter(exc)
def pause_writing(self):
"""Вызывается, когда буфер низкоуровневого транспорта превышает
верхнюю отметку.
"""
self._app_protocol.pause_writing()
def resume_writing(self):
"""Вызывается, когда буфер низкоуровневого транспорта опускается ниже
нижней отметки.
"""
self._app_protocol.resume_writing()
def data_received(self, data):
"""Вызывается при получении некоторых SSL-данных.
Аргументом является объект bytes.
"""
if self._sslpipe is None:
# транспорт закрывается, sslpipe уничтожен
return
try:
ssldata, appdata = self._sslpipe.feed_ssldata(data)
except ssl.SSLError as e:
if self._loop.get_debug():
logger.warning('%r: SSL error %s (reason %s)',
self, e.errno, e.reason)
self._abort()
return
for chunk in ssldata:
self._transport.write(chunk)
for chunk in appdata:
if chunk:
self._app_protocol.data_received(chunk)
else:
self._start_shutdown()
break
def eof_received(self):
"""Вызывается, когда другой конец низкоуровневого потока данных
наполовину закрыт.
Если этот метод возвращает ложное значение (включая None), транспорт
закроет себя. Если он возвращает истинное значение, закрытие
транспорта остаётся на усмотрение протокола.
"""
try:
if self._loop.get_debug():
logger.debug("%r received EOF", self)
self._wakeup_waiter(ConnectionResetError)
if not self._in_handshake:
keep_open = self._app_protocol.eof_received()
if keep_open:
logger.warning('returning true from eof_received() '
'has no effect when using ssl')
finally:
self._transport.close()
def _get_extra_info(self, name, default=None):
if name in self._extra:
return self._extra[name]
elif self._transport is not None:
return self._transport.get_extra_info(name, default)
else:
return default
def _start_shutdown(self):
if self._in_shutdown:
return
if self._in_handshake:
self._abort()
else:
self._in_shutdown = True
self._write_appdata(b'')
def _write_appdata(self, data):
self._write_backlog.append((data, 0))
self._write_buffer_size += len(data)
self._process_write_backlog()
def _start_handshake(self):
if self._loop.get_debug():
logger.debug("%r starts SSL handshake", self)
self._handshake_start_time = self._loop.time()
else:
self._handshake_start_time = None
self._in_handshake = True
# (b'', 1) — это специальное значение в _process_write_backlog() для выполнения
# рукопожатия SSL
self._write_backlog.append((b'', 1))
self._process_write_backlog()
def _on_handshake_complete(self, handshake_exc):
self._in_handshake = False
sslobj = self._sslpipe.ssl_object
try:
if handshake_exc is not None:
raise handshake_exc
peercert = sslobj.getpeercert()
if not hasattr(self._sslcontext, 'check_hostname'):
# Проверить имя хоста, если запрошено; Python 3.4+ использует check_hostname
# и проверяет имя хоста в 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():
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._transport.close()
if isinstance(exc, Exception):
self._wakeup_waiter(exc)
return
else:
raise
if self._loop.get_debug():
dt = self._loop.time() - self._handshake_start_time
logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3)
# Добавить дополнительную информацию, которая становится доступной после рукопожатия.
self._extra.update(peercert=peercert,
cipher=sslobj.cipher(),
compression=sslobj.compression(),
ssl_object=sslobj,
)
if self._call_connection_made:
self._app_protocol.connection_made(self._app_transport)
self._wakeup_waiter()
self._session_established = True
# В случае, если transport.write() уже был вызван. Не вызывать
# немедленно _process_write_backlog(), а запланировать его:
# _on_handshake_complete() может быть вызван косвенно из
# _process_write_backlog(), и _process_write_backlog() не является
# реентерабельным.
self._loop.call_soon(self._process_write_backlog)
def _process_write_backlog(self):
# Попытаться продвинуться в работе с отложенной записью.
if self._transport is None or self._sslpipe is None:
return
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 = self._sslpipe.do_handshake(
self._on_handshake_complete)
offset = 1
else:
ssldata = self._sslpipe.shutdown(self._finalize)
offset = 1
for chunk in ssldata:
self._transport.write(chunk)
if offset < len(data):
self._write_backlog[0] = (data, offset)
# Короткая запись означает, что запись заблокирована ожиданием чтения
# Необходимо включить чтение, если оно приостановлено!
assert self._sslpipe.need_ssldata
if self._transport._paused:
self._transport.resume_reading()
break
# Целый фрагмент из отложенной записи был обработан. Мы можем
# удалить его и уменьшить текущий размер буфера.
del self._write_backlog[0]
self._write_buffer_size -= len(data)
except BaseException as exc:
if self._in_handshake:
# BaseExceptions будут повторно возбуждены в _on_handshake_complete.
self._on_handshake_complete(exc)
else:
self._fatal_error(exc, 'Fatal error on SSL transport')
if not isinstance(exc, Exception):
# BaseException
raise
def _fatal_error(self, exc, message='Fatal error on transport'):
# Должен вызываться только из обработчика исключений.
if isinstance(exc, base_events._FATAL_ERROR_IGNORE):
if self._loop.get_debug():
logger.debug("%r: %s", self, message, exc_info=True)
else:
self._loop.call_exception_handler({
'message': message,
'exception': exc,
'transport': self._transport,
'protocol': self,
})
if self._transport:
self._transport._force_close(exc)
def _finalize(self):
self._sslpipe = None
if self._transport is not None:
self._transport.close()
def _abort(self):
try:
if self._transport is not None:
self._transport.abort()
finally:
self._finalize()