Always use SSLTransport_Transport when ktls is not requested

HISTORY.rst

@@ -5,6 +5,11 @@ aiofastnet Release History
    :depth: 1
    :local:
 
+0.23.0
+------------------
+
+* Simplify SSL transport layer, always use SSLTransport_Transport when KTLS is not requested
+
 0.22.0
 ------------------
 

aiofastnet/__init__.py

@@ -50,5 +50,5 @@
     ))
 
 
-__version__ = "0.22.0"
+__version__ = "0.23.0"
 __author__ = "Taras Kozlov"

aiofastnet/api_utils.py

@@ -8,11 +8,14 @@
 
 import asyncio
 import errno
+import platform
+import re
 import socket
 import ssl
 import weakref
 from asyncio.trsock import TransportSocket
 from logging import getLogger
+from pathlib import Path
 from typing import Callable
 
 from . import constants, openssl_compat
@@ -61,6 +64,58 @@ def _ssl_needs_fallback_engine(sslcontext: ssl.SSLContext) -> bool:
     return openssl_compat.OPENSSL_DYN_LIBS is None or getattr(sslcontext, "_aiofastnet_force_fallback_ssl", False)
 
 
+def _linux_kernel_at_least(major: int, minor: int) -> bool:
+    if platform.system() != "Linux":
+        return False
+
+    match = re.match(r"^(\d+)\.(\d+)", platform.release())
+    if match is None:
+        return False
+
+    current = tuple(map(int, match.groups()))
+    return current >= (major, minor)
+
+
+def _ktls_prerequisites_available() -> bool:
+    if not Path("/sys/module/tls").exists():
+        _logger.warning(
+            "Kernel TLS was requested but is unavailable because kernel module "
+            "'tls' is not loaded; load it with 'sudo modprobe tls'. "
+            "Falling back to memory BIO.")
+        return False
+
+    if not _linux_kernel_at_least(5, 1):
+        _logger.warning(
+            "Kernel TLS was requested but is unavailable because the Linux "
+            "kernel version is < 5.1. Falling back to memory BIO.")
+        return False
+
+    if ssl.OPENSSL_VERSION_INFO[:3] < (3, 0, 0):
+        _logger.warning(
+            "Kernel TLS was requested but is unavailable because OpenSSL "
+            "version is too old; OpenSSL >= 3.0 is required. "
+            "Falling back to memory BIO.")
+        if openssl_compat.OPENSSL_DYN_LIBS is not None:
+            _logger.warning("Loaded libssl: %s", openssl_compat.OPENSSL_DYN_LIBS.libssl)
+            _logger.warning("Loaded libcrypto: %s", openssl_compat.OPENSSL_DYN_LIBS.libcrypto)
+        return False
+
+    return True
+
+
+def _ssl_should_use_socket_bio(sslcontext: ssl.SSLContext) -> bool:
+    if _ssl_needs_fallback_engine(sslcontext):
+        return False
+
+    force_socket_bio = getattr(sslcontext, "_aiofastnet_force_socket_bio", False)
+    ktls_requested = (sslcontext.options & getattr(ssl, "OP_ENABLE_KTLS", 0)) != 0
+
+    # force_socket_bio is only used for testing, tests should not use it together with OP_ENABLE_KTLS
+    assert not (ktls_requested and force_socket_bio)
+
+    return force_socket_bio or (ktls_requested and _ktls_prerequisites_available())
+
+
 async def _create_connection_transport(
         loop: asyncio.AbstractEventLoop,
         sock: socket.socket,
@@ -128,32 +183,33 @@ def wrapped_protocol_factory():
         waiter = loop.create_future() if server is None else None
         if ssl:
             sslcontext = openssl_compat.create_transport_context(server_side, server_hostname) if isinstance(ssl, bool) else ssl
-            if _ssl_needs_fallback_engine(sslcontext):
-                transport = SSLTransport_Transport(
+            if _ssl_should_use_socket_bio(sslcontext):
+                transport = SSLTransport_Socket(
                     loop, protocol, sslcontext,
                     server_side,
                     ssl_handshake_timeout,
                     ssl_shutdown_timeout,
                     ssl_incoming_bio_size,
                     ssl_outgoing_bio_size,
+                    sock,
                     waiter=waiter,
                     server_hostname=server_hostname,
                     server=server
                 )
-                SocketTransport(loop, sock, transport.get_tls_protocol())
             else:
-                transport = SSLTransport_Socket(
+                transport = SSLTransport_Transport(
                     loop, protocol, sslcontext,
                     server_side,
                     ssl_handshake_timeout,
                     ssl_shutdown_timeout,
                     ssl_incoming_bio_size,
                     ssl_outgoing_bio_size,
-                    sock,
                     waiter=waiter,
                     server_hostname=server_hostname,
                     server=server
                 )
+                SocketTransport(loop, sock, transport.get_tls_protocol())
+
         else:
             transport = SocketTransport(loop, sock, protocol,
                                         waiter=waiter, server=server)

aiofastnet/ssl_engine_direct.pyx

@@ -100,13 +100,10 @@ from libc.limits cimport INT_MAX
 from libc.stdint cimport uint64_t
 
 import os
-import platform
-import re
 import ssl
 import sys
 import tempfile
 import logging
-from pathlib import Path
 
 cdef object _logger = logging.getLogger('aiofastnet.ssl')
 
@@ -115,44 +112,6 @@ def _set_sslobject_init_test_hook():
     pass
 
 
-def _linux_kernel_at_least(major: int, minor: int) -> bool:
-    if platform.system() != "Linux":
-        return False
-
-    match = re.match(r"^(\d+)\.(\d+)", platform.release())
-    if match is None:
-        return False
-
-    current = tuple(map(int, match.groups()))
-    return current >= (major, minor)
-
-
-def _ktls_prerequisites_available() -> bool:
-    if not Path("/sys/module/tls").exists():
-        _logger.warning(
-            "Kernel TLS was requested but is unavailable because kernel module "
-            "'tls' is not loaded; load it with 'sudo modprobe tls'. "
-            "Falling back to memory BIO.")
-        return False
-
-    if not _linux_kernel_at_least(5, 1):
-        _logger.warning(
-            "Kernel TLS was requested but is unavailable because the Linux "
-            "kernel version is < 5.1. Falling back to memory BIO.")
-        return False
-
-    if ssl.OPENSSL_VERSION_INFO[:3] < (3, 0, 0):
-        _logger.warning(
-            "Kernel TLS was requested but is unavailable because OpenSSL "
-            "version is too old; OpenSSL >= 3.0 is required. "
-            "Falling back to memory BIO.")
-        _logger.warning("Loaded libssl: %s", OPENSSL_DYN_LIBS.libssl)
-        _logger.warning("Loaded libcrypto: %s", OPENSSL_DYN_LIBS.libcrypto)
-        return False
-
-    return True
-
-
 cdef _init_openssl():
     assert OPENSSL_DYN_LIBS is not None
 
@@ -244,22 +203,9 @@ cdef class SSLEngineDirect(SSLEngine):
         self.outgoing = NULL
         self.ssl = NULL
 
-        cdef bint force_socket_bio = getattr(ssl_context, "_aiofastnet_force_socket_bio", False)
-
-        # force_socket_bio is only used for testing, tests should not use it together with OP_ENABLE_KTLS
-        assert not self.ktls_requested or (self.ktls_requested and not force_socket_bio)
-
-        cdef bint ktls_prerequisites_available = (
-            _ktls_prerequisites_available() if self.ktls_requested else False
-        )
         cdef bint enable_ktls = (
-            SSL_set_options_available() and
-            self.ktls_requested and
-            ktls_prerequisites_available
-        )
-        cdef bint use_socket_bio = (
             sock is not None and
-            (force_socket_bio or enable_ktls)
+            self.ktls_requested
         )
 
         cdef BIO* incoming = NULL
@@ -276,7 +222,7 @@ cdef class SSLEngineDirect(SSLEngine):
             if sys.version_info[:2] < (3, 10):
                 SSL_clear_options(self.ssl, SSL_OP_IGNORE_UNEXPECTED_EOF)
 
-            if use_socket_bio:
+            if sock is not None:
                 if SSL_set_fd(self.ssl, sock.fileno()) != 1:
                     raise ssl.SSLError("SSL_set_fd failed")
                 if enable_ktls:
@@ -847,4 +793,3 @@ cdef class SSLEngineDirect(SSLEngine):
         finally:
             if path:
                 os.unlink(path)
-

aiofastnet/ssl_transport.pyx

@@ -24,8 +24,6 @@ from .utils cimport (
     aiofn_validate_buffer,
     aiofn_maybe_copy_buffer,
     aiofn_maybe_copy_buffer_tail,
-    aiofn_recv,
-    aiofn_send,
     aiofn_allocate_bytes,
     aiofn_finalize_bytes,
     aiofn_set_nodelay,
@@ -1091,6 +1089,9 @@ cdef class SSLTransport_Socket(SSLTransportBase):
                                   server,
                                   sock)
 
+        assert not self._ssl_engine.ssl_incoming_use_membio()
+        assert not self._ssl_engine.ssl_outgoing_use_membio()
+
         self._extra['socket'] = TransportSocket(sock)
         try:
             self._extra['sockname'] = sock.getsockname()
@@ -1164,10 +1165,8 @@ cdef class SSLTransport_Socket(SSLTransportBase):
             _logger.debug("%r: reading resumed by user", self)
         self._loop.add_reader(self._sock_fd_obj, self._read_ready)
 
-        # We need to also manually schedule _read_ready event because there
-        # might be some leftover data in incoming BIO or openssl internal
-        # read buffer. We can't rely only on _loop.add_reader, because if
-        # socket has no data to read then we will get stuck.
+        # OpenSSL may have already buffered decrypted data, so socket
+        # readability alone is not sufficient to resume processing.
         if self._state in (SSLProtocolState.WRAPPED, SSLProtocolState.FLUSHING, SSLProtocolState.SHUTDOWN):
             self._loop.call_soon(self._read_ready)
 
@@ -1188,51 +1187,14 @@ cdef class SSLTransport_Socket(SSLTransportBase):
         return total
 
     cdef bint _flush_outgoing_bio(self) except -1:
-        """
-        Writes raw data to socket for outgoing BIO. 
-        Returns True if write operations can continue.
-        True is also returned if memory bio is not used, is such case _flush_outgoing_bio is no-op. 
-        """
-        if not self._ssl_engine.ssl_outgoing_use_membio():
-            return True
-
-        if self._write_had_eagain:
-            return False
-
-        cdef:
-            char* ptr
-            long sz
-            Py_ssize_t bytes_sent
-
-        while True:
-            sz = self._ssl_engine.outgoing_bio_get_data(&ptr)
-            if sz == 0:
-                return True
-
-            bytes_sent = aiofn_send(self._sock_fd, ptr, sz)
-            if unlikely(self._is_debug):
-                _logger.debug("%r: aiofn_send(...,len=%d)=%d", self, sz, bytes_sent)
-
-            if bytes_sent < 0:
-                self._ensure_writer()
-                return False
-
-            self._ssl_engine.outgoing_bio_consume(bytes_sent)
-            if bytes_sent == sz:
-                return True
-
-            ptr += bytes_sent
-            sz -= bytes_sent
+        return True
 
     cdef bint _should_retry_after_want_write(self) except -1:
         """
         Return True if we should retry the last operation after we got SSL_ERROR_WANT_WRITE
         """
-        if self._ssl_engine.ssl_outgoing_use_membio():
-            return self._flush_outgoing_bio()
-        else:
-            self._ensure_writer()
-            return False
+        self._ensure_writer()
+        return False
 
     cdef bint _should_flush_outgoing_after_read(self) except -1:
         return not self._write_ready_registered
@@ -1270,8 +1232,6 @@ cdef class SSLTransport_Socket(SSLTransportBase):
         self._write_had_eagain = False
 
         try:
-            self._flush_outgoing_bio()
-
             if self._state == SSLProtocolState.DO_HANDSHAKE:
                 self._do_handshake()
             elif self._state == SSLProtocolState.WRAPPED:
@@ -1327,31 +1287,8 @@ cdef class SSLTransport_Socket(SSLTransportBase):
         if self._connection_lost_scheduled:
             return
 
-        cdef:
-            char* buf_ptr
-            Py_ssize_t buf_len
-            Py_ssize_t bytes_read
-
         try:
-            if self._ssl_engine.ssl_incoming_use_membio():
-                while not self._read_paused:
-                    self._ssl_engine.incoming_bio_get_write_buf(&buf_ptr, &buf_len)
-                    bytes_read = aiofn_recv(self._sock_fd, buf_ptr, buf_len)
-
-                    if unlikely(self._is_debug):
-                        _logger.debug("%r: aiofn_recv(...,len=%d)=%d", self, buf_len, bytes_read)
-
-                    if bytes_read == -1:  # without exception this means EGAIN
-                        return
-
-                    if unlikely(bytes_read == 0):
-                        self._process_eof()
-                        return
-
-                    self._ssl_engine.incoming_bio_produce(bytes_read)
-                    self._incoming_bio_updated()
-            else:
-                self._incoming_bio_updated()
+            self._incoming_bio_updated()
         except:
             self._handle_error("Error occurred during read")
 
@@ -1478,6 +1415,9 @@ cdef class SSLTransport_Transport(SSLTransportBase):
                                   server_hostname,
                                   server)
 
+        assert self._ssl_engine.ssl_incoming_use_membio()
+        assert self._ssl_engine.ssl_outgoing_use_membio()
+
         self._transport = None
         self._is_aiofn_transport = False
         self._is_direct_engine = SSLEngineDirect is not None and isinstance(self._ssl_engine, SSLEngineDirect)