PYTHON-1341 Impl of client-side column-level encryption/decryption (d…

…atastax#1150)

.travis.yml

@@ -0,0 +1,30 @@
+dist: xenial
+sudo: false
+
+language: python
+python:
+  - "3.7"
+  - "3.8"
+
+env:
+  - CASS_DRIVER_NO_CYTHON=1
+
+addons:
+  apt:
+    packages:
+    - build-essential
+    - python-dev
+    - pypy-dev
+    - libc-ares-dev
+    - libev4
+    - libev-dev
+
+install:
+  - pip install --upgrade setuptools importlib-metadata
+  - pip install tox-travis
+  - if [[ $TRAVIS_PYTHON_VERSION != pypy3.5 ]]; then pip install lz4; fi
+
+script:
+  - tox
+  - tox -e gevent_loop
+  - tox -e eventlet_loop

Jenkinsfile

@@ -57,7 +57,7 @@ matrices = [
   "SMOKE": [
     "SERVER": ['3.11', '4.0', 'dse-6.8.30'],
     "RUNTIME": ['3.7.7', '3.8.3'],
-    "CYTHON": ["False"]
+    "CYTHON": ["True", "False"]
   ]
 ]
 

cassandra/cluster.py

@@ -1033,6 +1033,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.
+    """
+
     @property
     def schema_metadata_enabled(self):
         """
@@ -1148,7 +1154,9 @@ def __init__(self,
                  client_id=None,
                  cloud=None,
                  scylla_cloud=None,
-                 shard_aware_options=None):
+                 shard_aware_options=None,
+                 column_encryption_policy=None,
+                 ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
         extablishing connection pools or refreshing metadata.
@@ -1225,6 +1233,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)
 
@@ -2647,6 +2658,12 @@ def __init__(self, cluster, hosts, keyspace=None):
 
         self.encoder = Encoder()
 
+        if self.cluster.column_encryption_policy is not None:
+            try:
+                self.client_protocol_handler.column_encryption_policy = self.cluster.column_encryption_policy
+            except AttributeError:
+                log.info("Unable to set column encryption policy for session")
+
         # create connection pools in parallel
         self._initial_connect_futures = set()
         for host in hosts:
@@ -3186,7 +3203,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/obj_parser.pyx

@@ -17,9 +17,12 @@ include "ioutils.pyx"
 from cassandra import DriverException
 from cassandra.bytesio cimport BytesIOReader
 from cassandra.deserializers cimport Deserializer, from_binary
+from cassandra.deserializers import find_deserializer
 from cassandra.parsing cimport ParseDesc, ColumnParser, RowParser
 from cassandra.tuple cimport tuple_new, tuple_set
 
+from cpython.bytes cimport PyBytes_AsStringAndSize
+
 
 cdef class ListParser(ColumnParser):
     """Decode a ResultMessage into a list of tuples (or other objects)"""
@@ -58,18 +61,29 @@ cdef class TupleRowParser(RowParser):
         assert desc.rowsize >= 0
 
         cdef Buffer buf
+        cdef Buffer newbuf
         cdef Py_ssize_t i, rowsize = desc.rowsize
         cdef Deserializer deserializer
         cdef tuple res = tuple_new(desc.rowsize)
 
+        ce_policy = desc.column_encryption_policy
         for i in range(rowsize):
             # Read the next few bytes
             get_buf(reader, &buf)
 
             # Deserialize bytes to python object
             deserializer = desc.deserializers[i]
+            coldesc = desc.coldescs[i]
+            uses_ce = ce_policy and ce_policy.contains_column(coldesc)
             try:
-                val = from_binary(deserializer, &buf, desc.protocol_version)
+                if uses_ce:
+                    col_type = ce_policy.column_type(coldesc)
+                    decrypted_bytes = ce_policy.decrypt(coldesc, to_bytes(&buf))
+                    PyBytes_AsStringAndSize(decrypted_bytes, &newbuf.ptr, &newbuf.size)
+                    deserializer = find_deserializer(ce_policy.column_type(coldesc))
+                    val = from_binary(deserializer, &newbuf, desc.protocol_version)
+                else:
+                    val = from_binary(deserializer, &buf, desc.protocol_version)
             except Exception as e:
                 raise DriverException('Failed decoding result column "%s" of type %s: %s' % (desc.colnames[i],
                                                                                              desc.coltypes[i].cql_parameterized_type(),

cassandra/parsing.pxd

@@ -18,6 +18,8 @@ from cassandra.deserializers cimport Deserializer
 cdef class ParseDesc:
     cdef public object colnames
     cdef public object coltypes
+    cdef public object column_encryption_policy
+    cdef public list coldescs
     cdef Deserializer[::1] deserializers
     cdef public int protocol_version
     cdef Py_ssize_t rowsize

cassandra/parsing.pyx

@@ -19,9 +19,11 @@ Module containing the definitions and declarations (parsing.pxd) for parsers.
 cdef class ParseDesc:
     """Description of what structure to parse"""
 
-    def __init__(self, colnames, coltypes, deserializers, protocol_version):
+    def __init__(self, colnames, coltypes, column_encryption_policy, coldescs, deserializers, protocol_version):
         self.colnames = colnames
         self.coltypes = coltypes
+        self.column_encryption_policy = column_encryption_policy
+        self.coldescs = coldescs
         self.deserializers = deserializers
         self.protocol_version = protocol_version
         self.rowsize = len(colnames)