Merge pull request #364 from dkropachev/dk/sync-with-upstream-3.28.0

Sync with upstream 3.28.0

Author: dkropachev

.gitignore

@@ -3,6 +3,7 @@
 *.swo
 *.so
 *.egg
+*.eggs
 *.egg-info
 *.attr
 .tox

CHANGELOG.rst

@@ -1,3 +1,33 @@
+3.28.0
+======
+June 5, 2023
+
+Features
+--------
+* Add support for vector type (PYTHON-1352)
+* Cryptography module is now an optional dependency (PYTHON-1351)
+
+Bug Fixes
+---------
+* Store IV along with encrypted text when using column-level encryption (PYTHON-1350)
+* Create session-specific protocol handlers to contain session-specific CLE policies (PYTHON-1356)
+
+Others
+------
+* Use Cython for smoke builds (PYTHON-1343)
+* Don't fail when inserting UDTs with prepared queries with some missing fields (PR 1151)
+* Convert print statement to function in docs (PR 1157)
+* Update comment for retry policy (DOC-3278)
+* Added error handling blog reference (DOC-2813)
+
+3.27.0
+======
+May 1, 2023
+
+Features
+--------
+* Add support for client-side encryption (PYTHON-1341)
+
 3.26.0
 ======
 March 13, 2023
@@ -17,7 +47,7 @@ Others
 * Fix deprecation warning in query tracing (PR 1103)
 * Remove mutable default values from some tests (PR 1116)
 * Remove dependency on unittest2 (PYTHON-1289)
-* Fix deprecation warnings for asyncio.coroutine annotation in asyncioreactor (PYTTHON-1290)
+* Fix deprecation warnings for asyncio.coroutine annotation in asyncioreactor (PYTHON-1290)
 * Fix typos in source files (PR 1126)
 * HostFilterPolicyInitTest fix for Python 3.11 (PR 1131)
 * Fix for DontPrepareOnIgnoredHostsTest (PYTHON-1287)

README.rst

@@ -69,6 +69,10 @@ Contributing
 ------------
 See `CONTRIBUTING <https://github.com/scylladb/python-driver/blob/master/CONTRIBUTING.rst>`_.
 
+Error Handling
+------------
+While originally written for the Java driver, users may reference the `Cassandra error handling done right blog <https://www.datastax.com/blog/cassandra-error-handling-done-right>`_ for resolving error handling scenarios with Apache Cassandra.
+
 Reporting Problems
 ------------------
 Please report any bugs and make any feature requests by clicking the New Issue button in 

cassandra/__init__.py

@@ -23,7 +23,7 @@ def emit(self, record):
 
 logging.getLogger('cassandra').addHandler(NullHandler())
 
-__version_info__ = (3, 26, 9)
+__version_info__ = (3, 28, 0)
 __version__ = '.'.join(map(str, __version_info__))
 
 

cassandra/cluster.py

@@ -1015,7 +1015,7 @@ def default_retry_policy(self, policy):
     cloud = None
     """
     A dict of the cloud configuration. Example::
-        
+
         {
             # path to the secure connect bundle
             'secure_connect_bundle': '/path/to/secure-connect-dbname.zip',
@@ -1034,6 +1034,12 @@ def default_retry_policy(self, policy):
     or to disable the shardaware port (advanced shardaware)
     """
 
+    column_encryption_policy = None
+    """
+    An instance of :class:`cassandra.policies.ColumnEncryptionPolicy` specifying encryption materials to be
+    used for columns in this cluster.
+    """
+
     metadata_request_timeout = datetime.timedelta(seconds=2)
     """
     Timeout for all queries used by driver it self.
@@ -1157,6 +1163,7 @@ def __init__(self,
                  scylla_cloud=None,
                  shard_aware_options=None,
                  metadata_request_timeout=None,
+                 column_encryption_policy=None,
                  ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
@@ -1234,6 +1241,9 @@ def __init__(self,
 
         self.port = port
 
+        if column_encryption_policy is not None:
+            self.column_encryption_policy = column_encryption_policy
+
         self.endpoint_factory = endpoint_factory or DefaultEndPointFactory(port=self.port)
         self.endpoint_factory.configure(self)
 
@@ -1533,7 +1543,7 @@ def __init__(self, street, zipcode):
             # results will include Address instances
             results = session.execute("SELECT * FROM users")
             row = results[0]
-            print row.id, row.location.street, row.location.zipcode
+            print(row.id, row.location.street, row.location.zipcode)
 
         """
         if self.protocol_version < 3:
@@ -2678,6 +2688,17 @@ def __init__(self, cluster, hosts, keyspace=None):
         self.session_id = uuid.uuid4()
         self._graph_paging_available = self._check_graph_paging_available()
 
+        if self.cluster.column_encryption_policy is not None:
+            try:
+                self.client_protocol_handler = type(
+                    str(self.session_id) + "-ProtocolHandler",
+                    (ProtocolHandler,),
+                    {"column_encryption_policy": self.cluster.column_encryption_policy})
+            except AttributeError:
+                log.info("Unable to set column encryption policy for session")
+            raise Exception(
+                "column_encryption_policy is temporary disabled, until https://github.com/scylladb/python-driver/issues/365 is sorted out")
+
         if self.cluster.monitor_reporting_enabled:
             cc_host = self.cluster.get_control_connection_host()
             valid_insights_version = (cc_host and version_supports_insights(cc_host.dse_version))
@@ -3197,7 +3218,7 @@ def prepare(self, query, custom_payload=None, keyspace=None):
         prepared_keyspace = keyspace if keyspace else None
         prepared_statement = PreparedStatement.from_message(
             response.query_id, response.bind_metadata, response.pk_indexes, self.cluster.metadata, query, prepared_keyspace,
-            self._protocol_version, response.column_metadata, response.result_metadata_id)
+            self._protocol_version, response.column_metadata, response.result_metadata_id, self.cluster.column_encryption_policy)
         prepared_statement.custom_payload = future.custom_payload
 
         self.cluster.add_prepared(response.query_id, prepared_statement)

cassandra/column_encryption/_policies.py

@@ -0,0 +1,139 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from collections import namedtuple
+from functools import lru_cache
+
+import logging
+import os
+
+log = logging.getLogger(__name__)
+
+from cassandra.cqltypes import _cqltypes
+from cassandra.policies import ColumnEncryptionPolicy
+
+from cryptography.hazmat.primitives import padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
+AES256_BLOCK_SIZE = 128
+AES256_BLOCK_SIZE_BYTES = int(AES256_BLOCK_SIZE / 8)
+AES256_KEY_SIZE = 256
+AES256_KEY_SIZE_BYTES = int(AES256_KEY_SIZE / 8)
+
+ColData = namedtuple('ColData', ['key','type'])
+
+class AES256ColumnEncryptionPolicy(ColumnEncryptionPolicy):
+
+    # Fix block cipher mode for now.  IV size is a function of block cipher used
+    # so fixing this avoids (possibly unnecessary) validation logic here.
+    mode = modes.CBC
+
+    # "iv" param here expects a bytearray that's the same size as the block
+    # size for AES-256 (128 bits or 16 bytes).  If none is provided a new one
+    # will be randomly generated, but in this case the IV should be recorded and
+    # preserved or else you will not be able to decrypt any data encrypted by this
+    # policy.
+    def __init__(self, iv=None):
+
+        # CBC uses an IV that's the same size as the block size
+        #
+        # Avoid defining IV with a default arg in order to stay away from
+        # any issues around the caching of default args
+        self.iv = iv
+        if self.iv:
+            if not len(self.iv) == AES256_BLOCK_SIZE_BYTES:
+                raise ValueError("This policy uses AES-256 with CBC mode and therefore expects a 128-bit initialization vector")
+        else:
+            self.iv = os.urandom(AES256_BLOCK_SIZE_BYTES)
+
+        # ColData for a given ColDesc is always preserved.  We only create a Cipher
+        # when there's an actual need to for a given ColDesc
+        self.coldata = {}
+        self.ciphers = {}
+
+    def encrypt(self, coldesc, obj_bytes):
+
+        # AES256 has a 128-bit block size so if the input bytes don't align perfectly on
+        # those blocks we have to pad them.  There's plenty of room for optimization here:
+        #
+        # * Instances of the PKCS7 padder should be managed in a bounded pool
+        # * It would be nice if we could get a flag from encrypted data to indicate
+        #   whether it was padded or not
+        #   * Might be able to make this happen with a leading block of flags in encrypted data
+        padder = padding.PKCS7(AES256_BLOCK_SIZE).padder()
+        padded_bytes = padder.update(obj_bytes) + padder.finalize()
+
+        cipher = self._get_cipher(coldesc)
+        encryptor = cipher.encryptor()
+        return self.iv + encryptor.update(padded_bytes) + encryptor.finalize()
+
+    def decrypt(self, coldesc, bytes):
+
+        iv = bytes[:AES256_BLOCK_SIZE_BYTES]
+        encrypted_bytes = bytes[AES256_BLOCK_SIZE_BYTES:]
+        cipher = self._get_cipher(coldesc, iv=iv)
+        decryptor = cipher.decryptor()
+        padded_bytes = decryptor.update(encrypted_bytes) + decryptor.finalize()
+
+        unpadder = padding.PKCS7(AES256_BLOCK_SIZE).unpadder()
+        return unpadder.update(padded_bytes) + unpadder.finalize()
+
+    def add_column(self, coldesc, key, type):
+
+        if not coldesc:
+            raise ValueError("ColDesc supplied to add_column cannot be None")
+        if not key:
+            raise ValueError("Key supplied to add_column cannot be None")
+        if not type:
+            raise ValueError("Type supplied to add_column cannot be None")
+        if type not in _cqltypes.keys():
+            raise ValueError("Type %s is not a supported type".format(type))
+        if not len(key) == AES256_KEY_SIZE_BYTES:
+            raise ValueError("AES256 column encryption policy expects a 256-bit encryption key")
+        self.coldata[coldesc] = ColData(key, _cqltypes[type])
+
+    def contains_column(self, coldesc):
+        return coldesc in self.coldata
+
+    def encode_and_encrypt(self, coldesc, obj):
+        if not coldesc:
+            raise ValueError("ColDesc supplied to encode_and_encrypt cannot be None")
+        if not obj:
+            raise ValueError("Object supplied to encode_and_encrypt cannot be None")
+        coldata = self.coldata.get(coldesc)
+        if not coldata:
+            raise ValueError("Could not find ColData for ColDesc %s".format(coldesc))
+        return self.encrypt(coldesc, coldata.type.serialize(obj, None))
+
+    def cache_info(self):
+        return AES256ColumnEncryptionPolicy._build_cipher.cache_info()
+
+    def column_type(self, coldesc):
+        return self.coldata[coldesc].type
+
+    def _get_cipher(self, coldesc, iv=None):
+        """
+        Access relevant state from this instance necessary to create a Cipher and then get one,
+        hopefully returning a cached instance if we've already done so (and it hasn't been evicted)
+        """
+        try:
+            coldata = self.coldata[coldesc]
+            return AES256ColumnEncryptionPolicy._build_cipher(coldata.key, iv or self.iv)
+        except KeyError:
+            raise ValueError("Could not find column {}".format(coldesc))
+
+    # Explicitly use a class method here to avoid caching self
+    @lru_cache(maxsize=128)
+    def _build_cipher(key, iv):
+        return Cipher(algorithms.AES256(key), AES256ColumnEncryptionPolicy.mode(iv))

cassandra/column_encryption/policies.py

@@ -0,0 +1,20 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+try:
+    import cryptography
+    from cassandra.column_encryption._policies import *
+except ImportError:
+    # Cryptography is not installed
+    pass

cassandra/cqlengine/query.py

@@ -285,15 +285,15 @@ class ContextQuery(object):
 
             with ContextQuery(Automobile, keyspace='test2') as A:
                 A.objects.create(manufacturer='honda', year=2008, model='civic')
-                print len(A.objects.all())  # 1 result
+                print(len(A.objects.all()))  # 1 result
 
             with ContextQuery(Automobile, keyspace='test4') as A:
-                print len(A.objects.all())  # 0 result
+                print(len(A.objects.all()))  # 0 result
 
             # Multiple models
             with ContextQuery(Automobile, Automobile2, connection='cluster2') as (A, A2):
-                print len(A.objects.all())
-                print len(A2.objects.all())
+                print(len(A.objects.all()))
+                print(len(A2.objects.all()))
 
     """
 
@@ -808,11 +808,11 @@ class Comment(Model):
 
             print("Normal")
             for comment in Comment.objects(photo_id=u):
-                print comment.comment_id
+                print(comment.comment_id)
 
             print("Reversed")
             for comment in Comment.objects(photo_id=u).order_by("-comment_id"):
-                print comment.comment_id
+                print(comment.comment_id)
         """
         if len(colnames) == 0:
             clone = copy.deepcopy(self)