Skip to content

CASSPYTHON-13: Remove eventlet, gevent and twisted event loops #969

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
mykaul wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

CASSPYTHON-13: Remove eventlet, gevent and twisted event loops #969

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c532d23
CASSPYTHON-13: Remove eventlet, gevent and twisted event loops
brettabamonte May 24, 2026
e97d2dd
CASSPYTHON-13: Remove stale EventletConnection reference in _create_t…
mykaul Aug 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions CONTRIBUTING.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,7 @@ Running Unit Tests
Unit tests can be run like so::

uv run pytest tests/unit
EVENT_LOOP_MANAGER=gevent uv run pytest tests/unit/io/test_geventreactor.py
EVENT_LOOP_MANAGER=eventlet uv run pytest tests/unit/io/test_eventletreactor.py
EVENT_LOOP_MANAGER=asyncio uv run pytest tests/unit/io/test_asyncioreactor.py

You can run a specific test method like so::

Expand Down
16 changes: 0 additions & 16 deletions benchmarks/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,15 +65,6 @@
except (ImportError, SyntaxError):
pass

have_twisted = False
try:
from cassandra.io.twistedreactor import TwistedConnection
have_twisted = True
supported_reactors.append(TwistedConnection)
except ImportError as exc:
log.exception("Error importing twisted")
pass

KEYSPACE = "testkeyspace" + str(int(time.time()))
TABLE = "testtable"

Expand Down Expand Up @@ -228,8 +219,6 @@ def parse_options():
help='only benchmark with asyncio connections')
parser.add_option('--libev-only', action='store_true', dest='libev_only',
help='only benchmark with libev connections')
parser.add_option('--twisted-only', action='store_true', dest='twisted_only',
help='only benchmark with Twisted connections')
parser.add_option('-m', '--metrics', action='store_true', dest='enable_metrics',
help='enable and print metrics for operations')
parser.add_option('-l', '--log-level', default='info',
Expand Down Expand Up @@ -269,11 +258,6 @@ def parse_options():
log.error("libev is not available")
sys.exit(1)
options.supported_reactors = [LibevConnection]
elif options.twisted_only:
if not have_twisted:
log.error("Twisted is not available")
sys.exit(1)
options.supported_reactors = [TwistedConnection]
else:
options.supported_reactors = supported_reactors
if not have_libev:
Expand Down
90 changes: 4 additions & 86 deletions cassandra/cluster.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,6 @@
import re
import queue
import socket
import sys
import time
from threading import Lock, RLock, Thread, Event
import uuid
Expand Down Expand Up @@ -99,57 +98,11 @@
from cassandra.datastax import cloud as dscloud
from cassandra.application_info import ApplicationInfoBase

try:
from cassandra.io.twistedreactor import TwistedConnection
except ImportError:
TwistedConnection = None

try:
from cassandra.io.eventletreactor import EventletConnection
except (ImportError, AttributeError):
# AttributeError was add for handling python 3.12 https://github.com/eventlet/eventlet/issues/812
# TODO: remove it when eventlet issue would be fixed
EventletConnection = None

try:
from weakref import WeakSet
except ImportError:
from cassandra.util import WeakSet # NOQA

def _is_gevent_monkey_patched():
if 'gevent.monkey' not in sys.modules:
return False
try:
import gevent.socket
return socket.socket is gevent.socket.socket # Another case related to PYTHON-1364
except (AttributeError, ImportError):
return False

def _try_gevent_import():
if _is_gevent_monkey_patched():
from cassandra.io.geventreactor import GeventConnection
return (GeventConnection,None)
else:
return (None,None)

def _is_eventlet_monkey_patched():
if 'eventlet.patcher' not in sys.modules:
return False
try:
import eventlet.patcher
return eventlet.patcher.is_monkey_patched('socket')
except (ImportError, AttributeError):
# AttributeError was add for handling python 3.12 https://github.com/eventlet/eventlet/issues/812
# TODO: remove it when eventlet issue would be fixed
return False

def _try_eventlet_import():
if _is_eventlet_monkey_patched():
from cassandra.io.eventletreactor import EventletConnection
return (EventletConnection,None)
else:
return (None,None)

def _try_libev_import():
try:
from cassandra.io.libevreactor import LibevConnection
Expand Down Expand Up @@ -178,7 +131,7 @@
excs.append(exc)
return (rv or import_result, excs)

conn_fns = (_try_gevent_import, _try_eventlet_import, _try_libev_import, _try_asyncore_import, _try_asyncio_import)
conn_fns = (_try_libev_import, _try_asyncore_import, _try_asyncio_import)
(conn_class, excs) = reduce(_connection_reduce_fn, conn_fns, (None,[]))
if not conn_class:
raise DependencyException("Exception loading connection class dependencies", excs)
Expand Down Expand Up @@ -945,19 +898,13 @@

* :class:`cassandra.io.asyncorereactor.AsyncoreConnection`
* :class:`cassandra.io.libevreactor.LibevConnection`
* :class:`cassandra.io.eventletreactor.EventletConnection` (requires monkey-patching - see doc for details)
* :class:`cassandra.io.geventreactor.GeventConnection` (requires monkey-patching - see doc for details)
* :class:`cassandra.io.twistedreactor.TwistedConnection`
* EXPERIMENTAL: :class:`cassandra.io.asyncioreactor.AsyncioConnection`

By default, ``AsyncoreConnection`` will be used, which uses
the ``asyncore`` module in the Python standard library.

If ``libev`` is installed, ``LibevConnection`` will be used instead.

If ``gevent`` or ``eventlet`` monkey-patching is detected, the corresponding
connection class will be used automatically.

``AsyncioConnection``, which uses the ``asyncio`` module in the Python
standard library, is also available, but currently experimental. Note that
it requires ``asyncio`` features that were only introduced in the 3.4 line
Expand Down Expand Up @@ -1302,9 +1249,7 @@
raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options "
"cannot be specified with a cloud configuration")

uses_twisted = TwistedConnection and issubclass(self.connection_class, TwistedConnection)
uses_eventlet = EventletConnection and issubclass(self.connection_class, EventletConnection)
cloud_config = dscloud.get_cloud_config(cloud, create_pyopenssl_context=uses_twisted or uses_eventlet)
cloud_config = dscloud.get_cloud_config(cloud)

ssl_context = cloud_config.ssl_context
ssl_options = {'check_hostname': True}
Expand Down Expand Up @@ -1602,39 +1547,12 @@

def _create_thread_pool_executor(self, **kwargs):
"""
Create a ThreadPoolExecutor for the cluster. In most cases, the built-in
`concurrent.futures.ThreadPoolExecutor` is used.

Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor`
to hang indefinitely. In that case, the user needs to have the `futurist`
package so we can use the `futurist.GreenThreadPoolExecutor` class instead.
Create a ThreadPoolExecutor for the cluster.

:param kwargs: All keyword args are passed to the ThreadPoolExecutor constructor.
:return: A ThreadPoolExecutor instance.
"""
tpe_class = ThreadPoolExecutor
if sys.version_info[0] >= 3 and sys.version_info[1] >= 7:
try:
from cassandra.io.eventletreactor import EventletConnection
is_eventlet = issubclass(self.connection_class, EventletConnection)
except:
# Eventlet is not available or can't be detected
return tpe_class(**kwargs)

if is_eventlet:
try:
from futurist import GreenThreadPoolExecutor
tpe_class = GreenThreadPoolExecutor
except ImportError:
# futurist is not available
raise ImportError(
("Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor` "
"to hang indefinitely. If you want to use the Eventlet reactor, you "
"need to install the `futurist` package to allow the driver to use "
"the GreenThreadPoolExecutor. See https://github.com/eventlet/eventlet/issues/508 "
"for more details."))

return tpe_class(**kwargs)
return ThreadPoolExecutor(**kwargs)

def register_user_type(self, keyspace, user_type, klass):
"""
Expand Down Expand Up @@ -4635,7 +4553,7 @@
self._scheduled_tasks.discard(task)
fn, args, kwargs = task
kwargs = dict(kwargs)
future = self._executor.submit(fn, *args, **kwargs)

Check failure on line 4556 in cassandra/cluster.py

View workflow job for this annotation

GitHub Actions / test asyncore (3.11) 

cannot schedule new futures after shutdown
future.add_done_callback(self._log_if_failed)
else:
self._queue.put_nowait((run_at, i, task))
Expand Down
5 changes: 1 addition & 4 deletions cassandra/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,10 +35,7 @@
from cassandra.client_routes import _ClientRoutesHandler
from cassandra.protocol_features import ProtocolFeatures

if 'gevent.monkey' in sys.modules:
from gevent.queue import Queue, Empty
else:
from queue import Queue, Empty # noqa
from queue import Queue, Empty # noqa

from cassandra import ConsistencyLevel, AuthenticationFailed, OperationTimedOut, ProtocolVersion
from cassandra.marshal import int32_pack
Expand Down
31 changes: 5 additions & 26 deletions cassandra/datastax/cloud/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,38 +75,36 @@ def from_dict(cls, d):
return c


def get_cloud_config(cloud_config, create_pyopenssl_context=False):
def get_cloud_config(cloud_config):
if not _HAS_SSL:
raise DriverException("A Python installation with SSL is required to connect to a cloud cluster.")

if 'secure_connect_bundle' not in cloud_config:
raise ValueError("The cloud config doesn't have a secure_connect_bundle specified.")

try:
config = read_cloud_config_from_zip(cloud_config, create_pyopenssl_context)
config = read_cloud_config_from_zip(cloud_config)
except BadZipFile:
raise ValueError("Unable to open the zip file for the cloud config. Check your secure connect bundle.")

config = read_metadata_info(config, cloud_config)
if create_pyopenssl_context:
config.ssl_context = config.pyopenssl_context
return config


def read_cloud_config_from_zip(cloud_config, create_pyopenssl_context):
def read_cloud_config_from_zip(cloud_config):
secure_bundle = cloud_config['secure_connect_bundle']
use_default_tempdir = cloud_config.get('use_default_tempdir', None)
with ZipFile(secure_bundle) as zipfile:
base_dir = tempfile.gettempdir() if use_default_tempdir else os.path.dirname(secure_bundle)
tmp_dir = tempfile.mkdtemp(dir=base_dir)
try:
zipfile.extractall(path=tmp_dir)
return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config, create_pyopenssl_context)
return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config)
Comment on lines 97 to +102

@coderabbitai coderabbitai Bot Aug 9, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔒 Security & Privacy | 🟠 Major | ⚡ Quick win

Block Zip Slip paths before extraction.

extractall() accepts archive members such as ../../target. A crafted secure-connect bundle can write outside tmp_dir with the driver process permissions. Validate each resolved member path is within tmp_dir before extraction. Add a regression test with a traversal entry.

Proposed fix 
-            zipfile.extractall(path=tmp_dir)
+            root = os.path.realpath(tmp_dir)
+            for member in zipfile.infolist():
+                target = os.path.realpath(os.path.join(root, member.filename))
+                if os.path.commonpath((root, target)) != root:
+                    raise ValueError("The secure connect bundle contains an unsafe path.")
+                zipfile.extract(member, root)
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
with ZipFile(secure_bundle) as zipfile:
base_dir = tempfile.gettempdir() if use_default_tempdir else os.path.dirname(secure_bundle)
tmp_dir = tempfile.mkdtemp(dir=base_dir)
try:
zipfile.extractall(path=tmp_dir)
return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config, create_pyopenssl_context)
return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config)
with ZipFile(secure_bundle) as zipfile:
base_dir = tempfile.gettempdir() if use_default_tempdir else os.path.dirname(secure_bundle)
tmp_dir = tempfile.mkdtemp(dir=base_dir)
try:
root = os.path.realpath(tmp_dir)
for member in zipfile.infolist():
target = os.path.realpath(os.path.join(root, member.filename))
if os.path.commonpath((root, target)) != root:
raise ValueError("The secure connect bundle contains an unsafe path.")
zipfile.extract(member, root)
return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config)
🧰 Tools
🪛 ast-grep (0.45.0)

[error] 100-100: Calling extractall() on a zipfile.ZipFile or tarfile archive without validating member paths lets a crafted entry (e.g. "../../etc/passwd") write outside the destination directory (Zip Slip). Validate each member resolves inside the target directory, or pass a safe filter (tarfile: filter="data" / tarfile.data_filter).
Context: zipfile.extractall(path=tmp_dir)
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').

(archive-extractall-path-traversal-python)

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@cassandra/datastax/cloud/__init__.py` around lines 97 - 102, Harden the
extraction flow in the secure-bundle handling code around ZipFile.extractall by
resolving every archive member destination and rejecting any path outside
tmp_dir before extraction. Preserve normal extraction and parse_cloud_config
behavior for safe entries, and add a regression test covering a traversal
archive member.

Source: Linters/SAST tools

finally:
shutil.rmtree(tmp_dir)


def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
def parse_cloud_config(path, cloud_config):
with open(path, 'r') as stream:
data = json.load(stream)

Expand All @@ -120,11 +118,7 @@ def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
ca_cert_location = os.path.join(config_dir, 'ca.crt')
cert_location = os.path.join(config_dir, 'cert')
key_location = os.path.join(config_dir, 'key')
# Regardless of if we create a pyopenssl context, we still need the builtin one
# to connect to the metadata service
config.ssl_context = _ssl_context_from_cert(ca_cert_location, cert_location, key_location)
if create_pyopenssl_context:
config.pyopenssl_context = _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location)

return config

Expand Down Expand Up @@ -175,18 +169,3 @@ def _ssl_context_from_cert(ca_cert_location, cert_location, key_location):

return ssl_context


def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
try:
from OpenSSL import SSL
except ImportError as e:
raise ImportError(
"PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops")\
.with_traceback(e.__traceback__)
ssl_context = SSL.Context(SSL.TLSv1_METHOD)
ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok)
ssl_context.use_certificate_file(cert_location)
ssl_context.use_privatekey_file(key_location)
ssl_context.load_verify_locations(ca_cert_location)

return ssl_context
6 changes: 1 addition & 5 deletions cassandra/datastax/insights/reporter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,11 +142,7 @@ def _get_startup_data(self):
cert_validation = None
try:
if self._session.cluster.ssl_context:
if isinstance(self._session.cluster.ssl_context, ssl.SSLContext):
cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
else: # pyopenssl
from OpenSSL import SSL
cert_validation = self._session.cluster.ssl_context.get_verify_mode() != SSL.VERIFY_NONE
cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
elif self._session.cluster.ssl_options:
cert_validation = self._session.cluster.ssl_options.get('cert_reqs') == ssl.CERT_REQUIRED
except Exception as e:
Expand Down
Loading
Loading