DRILL-4335: Apache Drill should support network encryption.

NOTE: This pull request provides support for on-wire encryption using SASL framework. Communication channel covered is:
      1) C++ Drill Client and Drillbit channel.

close apache/drill#809

14 files changed, 1194 insertions, 324 deletions

diff --git a/contrib/native/client/example/querySubmitter.cpp b/contrib/native/client/example/querySubmitter.cpp
index 599089786..47e55de2d 100644
--- a/contrib/native/client/example/querySubmitter.cpp
+++ b/contrib/native/client/example/querySubmitter.cpp
@@ -1,3 +1,4 @@
+
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -24,7 +25,7 @@
 #include <boost/algorithm/string/join.hpp>
 #include "drill/drillc.hpp"
 
-int nOptions=15;
+int nOptions=19;
 
 struct Option{
     char name[32];
@@ -45,7 +46,11 @@ struct Option{
     {"heartbeatFrequency", "Heartbeat frequency (second). Disabled if set to 0.", false},
     {"user", "Username", false},
     {"password", "Password", false},
-    {"saslPluginPath", "Path to where SASL plugins are installed", false}
+    {"saslPluginPath", "Path to where SASL plugins are installed", false},
+    {"service_host", "Service host for Kerberos", false},
+    {"service_name", "Service name for Kerberos", false},
+    {"auth", "Authentication mechanism to use", false},
+    {"sasl_encrypt", "Negotiate for encrypted connection", false}
 };
 
 std::map<std::string, std::string> qsOptionValues;
@@ -295,6 +300,10 @@ int main(int argc, char* argv[]) {
         std::string user=qsOptionValues["user"];
         std::string password=qsOptionValues["password"];
         std::string saslPluginPath=qsOptionValues["saslPluginPath"];
+        std::string sasl_encrypt=qsOptionValues["sasl_encrypt"];
+        std::string serviceHost=qsOptionValues["service_host"];
+        std::string serviceName=qsOptionValues["service_name"];
+        std::string auth=qsOptionValues["auth"];
 
         Drill::QueryType type;
 
@@ -371,6 +380,18 @@ int main(int argc, char* argv[]) {
         if(password.length()>0){
             props.setProperty(USERPROP_PASSWORD, password);
         }
+        if(sasl_encrypt.length()>0){
+            props.setProperty(USERPROP_SASL_ENCRYPT, sasl_encrypt);
+        }
+        if(serviceHost.length()>0){
+            props.setProperty(USERPROP_SERVICE_HOST, serviceHost);
+        }
+        if(serviceName.length()>0){
+            props.setProperty(USERPROP_SERVICE_NAME, serviceName);
+        }
+        if(auth.length()>0){
+            props.setProperty(USERPROP_AUTH_MECHANISM, auth);
+        }
 
         if(client.connect(connectStr.c_str(), &props)!=Drill::CONN_SUCCESS){
             std::cerr<< "Failed to connect with error: "<< client.getError() << " (Using:"<<connectStr<<")"<<std::endl;
diff --git a/contrib/native/client/readme.macos b/contrib/native/client/readme.macos
index 4785e87df..eee017ef5 100644
--- a/contrib/native/client/readme.macos
+++ b/contrib/native/client/readme.macos
@@ -35,6 +35,8 @@ Install Prerequisites
   or use brew to install 
   $> brew install cmake
 
+2.0) Install cppunit 
+  $> brew install cppunit 
 
 2.1) Install protobuf 2.5.0 (or higher)
   $> brew install protobuf
@@ -54,7 +56,7 @@ Install Prerequisites
 When changes have been introduced to the protocol module, you might need to refresh the protobuf C++ source files too.
   $> cd DRILL_DIR/contrib/native/client
   $> mkdir build
-  $> cd build && cmake3 -G "XCode" -D CMAKE_BUILD_TYPE=Debug ..
+  $> cd build && cmake -G "Xcode" -D CMAKE_BUILD_TYPE=Debug ..
   $> xcodebuild -project drillclient.xcodeproj -configuration ${BUILDTYPE} -target fixProtobufs
   $> xcodebuild -project drillclient.xcodeproj -configuration ${BUILDTYPE} -target cpProtobufs
 
@@ -64,7 +66,7 @@ Build drill client
 -------------------
   $> cd DRILL_DIR/contrib/native/client
   $> mkdir build
-  $> cd build && cmake3 -G "XCode" -D CMAKE_BUILD_TYPE=Debug ..
+  $> cd build && cmake -G "Xcode" -D CMAKE_BUILD_TYPE=Debug ..
   $> xcodebuild -project drillclient.xcodeproj -configuration ${BUILDTYPE} -target ALL_BUILD
 
 
diff --git a/contrib/native/client/src/clientlib/drillClient.cpp b/contrib/native/client/src/clientlib/drillClient.cpp
index 70002721c..8eb909b49 100644
--- a/contrib/native/client/src/clientlib/drillClient.cpp
+++ b/contrib/native/client/src/clientlib/drillClient.cpp
@@ -181,6 +181,7 @@ const std::map<std::string, uint32_t>  DrillUserProperties::USER_PROPERTIES=boos
     ( USERPROP_USESSL,      USERPROP_FLAGS_BOOLEAN|USERPROP_FLAGS_SSLPROP)
     ( USERPROP_FILEPATH,    USERPROP_FLAGS_STRING|USERPROP_FLAGS_SSLPROP|USERPROP_FLAGS_FILEPATH)
     ( USERPROP_FILENAME,    USERPROP_FLAGS_STRING|USERPROP_FLAGS_SSLPROP|USERPROP_FLAGS_FILENAME)
+    ( USERPROP_SASL_ENCRYPT,  USERPROP_FLAGS_STRING)
 ;
 
 bool DrillUserProperties::validate(std::string& err){
diff --git a/contrib/native/client/src/clientlib/drillClientImpl.cpp b/contrib/native/client/src/clientlib/drillClientImpl.cpp
index 30a354e47..0dee309a6 100644
--- a/contrib/native/client/src/clientlib/drillClientImpl.cpp
+++ b/contrib/native/client/src/clientlib/drillClientImpl.cpp
@@ -30,7 +30,6 @@
 #include <boost/lexical_cast.hpp>
 #include <boost/thread.hpp>
 
-
 #include "drill/drillClient.hpp"
 #include "drill/fieldmeta.hpp"
 #include "drill/recordBatch.hpp"
@@ -193,7 +192,7 @@ connectionStatus_t DrillClientImpl::sendHeartbeat(){
     boost::lock_guard<boost::mutex> prLock(this->m_prMutex);
     boost::lock_guard<boost::mutex> lock(m_dcMutex);
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Heartbeat sent." << std::endl;)
-    status=sendSync(heartbeatMsg);
+    status=sendSyncCommon(heartbeatMsg);
     status=status==CONN_SUCCESS?status:CONN_DEAD;
     //If the server sends responses to a heartbeat, we need to increment the pending requests counter.
     if(m_pendingRequests++==0){
@@ -233,18 +232,125 @@ void DrillClientImpl::Close() {
     shutdownSocket();
 }
 
+/*
+ * Write bytesToWrite length data bytes pointed by dataPtr. It handles EINTR error
+ * occurred during write_some sys call and does a retry on that.
+ *
+ * Parameters:
+ *      dataPtr      - in param   - Pointer to data bytes to write on socket.
+ *      bytesToWrite - in param   - Length of data bytes to write from dataPtr.
+ *      errorCode    -  out param - Error code set by boost.
+ */
+void DrillClientImpl::doWriteToSocket(const char* dataPtr, size_t bytesToWrite,
+                                        boost::system::error_code& errorCode) {
+    if(0 == bytesToWrite) {
+        return;
+    }
+
+    // Write all the bytes to socket. In case of error when all bytes are not successfully written
+    // proper errorCode will be set.
+    while(1) {
+        size_t bytesWritten = m_socket.write_some(boost::asio::buffer(dataPtr, bytesToWrite), errorCode);
+
+        // Update the state
+        bytesToWrite -= bytesWritten;
+        dataPtr += bytesWritten;
+
+        if(EINTR != errorCode.value()) break;
+
+        // Check if all the data is written then break from loop
+        if(0 == bytesToWrite) break;
+    }
+}
 
-connectionStatus_t DrillClientImpl::sendSync(rpc::OutBoundRpcMessage& msg){
+/*
+ * Common wrapper to take care of sending both plain or encrypted message. It creates a send buffer from an
+ * OutboundRPCMessage and then call the send handler pointing to either sendSyncPlain or sendSyncEncrypted
+ *
+ * Return:
+ *  connectionStatus_t  -   CONN_SUCCESS - In case of successful send
+ *                      -   CONN_FAILURE - In case of failure to send
+ */
+connectionStatus_t DrillClientImpl::sendSyncCommon(rpc::OutBoundRpcMessage& msg) {
     encode(m_wbuf, msg);
+    return (this->*m_fpCurrentSendHandler)();
+}
+
+/*
+ * Send handler for sending plain messages over wire
+ *
+ * Return:
+ *  connectionStatus_t  -   CONN_SUCCESS - In case of successful send
+ *                      -   CONN_FAILURE - In case of failure to send
+ */
+connectionStatus_t DrillClientImpl::sendSyncPlain(){
+
     boost::system::error_code ec;
-    size_t s=m_socket.write_some(boost::asio::buffer(m_wbuf), ec);
-    if(!ec && s!=0){
+    doWriteToSocket(reinterpret_cast<char*>(m_wbuf.data()), m_wbuf.size(), ec);
+
+    if(!ec) {
         return CONN_SUCCESS;
-    }else{
+    } else {
         return handleConnError(CONN_FAILURE, getMessage(ERR_CONN_WFAIL, ec.message().c_str()));
     }
 }
 
+/*
+ * Send handler for sending encrypted messages over wire. It encrypts the send buffer using wrap api provided by
+ * saslAuthenticatorImpl and then transmit the encrypted bytes over wire.
+ *
+ * Return:
+ *  connectionStatus_t  -   CONN_SUCCESS - In case of successful send
+ *                      -   CONN_FAILURE - In case of failure to send
+ */
+connectionStatus_t DrillClientImpl::sendSyncEncrypted() {
+
+    boost::system::error_code ec;
+
+    // Encoded message is encrypted into chunks of size <= WrapSizeLimit. Each encrypted chunk along with
+    // its encrypted length in network order (added by Cyrus-SASL plugin) is sent over wire.
+    const int wrapChunkSize = m_encryptionCtxt.getWrapSizeLimit();
+    int lengthToEncrypt = m_wbuf.size();
+
+    int currentChunkLen = std::min(wrapChunkSize, lengthToEncrypt);
+    uint32_t currentChunkOffset = 0;
+    std::stringstream errorMsg;
+
+    // Encrypt and send each chunk
+    while(lengthToEncrypt != 0) {
+        const char* wrappedChunk = NULL;
+        uint32_t wrappedLen = 0;
+        const int wrapResult = m_saslAuthenticator->wrap(reinterpret_cast<const char*>(m_wbuf.data() + currentChunkOffset),
+                                                   currentChunkLen, &wrappedChunk, wrappedLen);
+        if(SASL_OK != wrapResult) {
+            errorMsg << "Sasl wrap failed while encrypting chunk of length: " << currentChunkLen << " , EncodeError: "
+                     << wrapResult;
+            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::sendSyncEncrypted - " << errorMsg.str()
+                                              << " ,ChunkOffset: " << currentChunkOffset << ", Message Len: " << m_wbuf.size()
+                                              << ", Closing connection.";)
+            return handleConnError(CONN_FAILURE, getMessage(ERR_CONN_WFAIL, errorMsg.str().c_str()));
+        }
+
+        // Send the encrypted chunk.
+        doWriteToSocket(wrappedChunk, wrappedLen, ec);
+
+        if(ec) {
+            errorMsg << "Failure while sending encrypted chunk. Error: " << ec.message().c_str();
+            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::sendSyncEncrypted - " << errorMsg.str()
+                                              << ", Chunk Length: " << currentChunkLen << ", ChunkOffset:" << currentChunkOffset
+                                              << ", Message Len: " << m_wbuf.size() << ", Closing connection.";)
+            return handleConnError(CONN_FAILURE, getMessage(ERR_CONN_WFAIL, errorMsg.str().c_str()));
+        }
+
+        // Update variables after sending each encrypted chunk
+        lengthToEncrypt -= currentChunkLen;
+        currentChunkOffset += currentChunkLen;
+        currentChunkLen = std::min(wrapChunkSize, lengthToEncrypt);
+    }
+
+    return CONN_SUCCESS;
+}
+
 connectionStatus_t DrillClientImpl::recvHandshake(){
     if(m_rbuf==NULL){
         m_rbuf = Utils::allocateBuffer(MAX_SOCK_RD_BUFSIZE);
@@ -289,7 +395,41 @@ connectionStatus_t DrillClientImpl::recvHandshake(){
     return CONN_SUCCESS;
 }
 
-void DrillClientImpl::handleHandshake(ByteBuf_t _buf,
+/*
+ * Read bytesToRead length data bytes from socket into inBuf. It handles EINTR error
+ * occurred during read_some sys call and does a retry on that.
+ *
+ * Parameters:
+ *      inBuf        - out param  - Pointer to buffer to read data into from socket.
+ *      bytesToRead  - in param   - Length of data bytes to read from socket.
+ *      errorCode    - out param  - Error code set by boost.
+ */
+void DrillClientImpl::doReadFromSocket(ByteBuf_t inBuf, size_t bytesToRead,
+                                       boost::system::error_code& errorCode) {
+
+    // Check if bytesToRead is zero
+    if(0 == bytesToRead) {
+        return;
+    }
+
+    // Read all the bytes. In case when all the bytes were not read the proper
+    // errorCode will be set.
+    while(1){
+        size_t dataBytesRead = m_socket.read_some(boost::asio::buffer(inBuf, bytesToRead),
+                                           errorCode);
+        // Update the state
+        bytesToRead -= dataBytesRead;
+        inBuf += dataBytesRead;
+
+        // Check if errorCode is EINTR then just retry otherwise break from loop
+        if(EINTR != errorCode.value()) break;
+
+        // Check if all the data is read then break from loop
+        if(0 == bytesToRead) break;
+    }
+}
+
+void DrillClientImpl::handleHandshake(ByteBuf_t inBuf,
         const boost::system::error_code& err,
         size_t bytes_transferred) {
     boost::system::error_code error=err;
@@ -299,21 +439,23 @@ void DrillClientImpl::handleHandshake(ByteBuf_t _buf,
     if(!error){
         rpc::InBoundRpcMessage msg;
         uint32_t length = 0;
-        std::size_t bytes_read = rpc::lengthDecode(m_rbuf, length);
+        std::size_t bytes_read = rpcLengthDecode(m_rbuf, length);
         if(length>0){
-            size_t leftover = LEN_PREFIX_BUFLEN - bytes_read;
-            ByteBuf_t b=m_rbuf + LEN_PREFIX_BUFLEN;
-            size_t bytesToRead=length - leftover;
-            while(1){
-                size_t dataBytesRead=m_socket.read_some(
-                        boost::asio::buffer(b, bytesToRead),
-                        error);
-                if(err) break;
-                DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Handshake Message: actual bytes read = " << dataBytesRead << std::endl;)
-                if(dataBytesRead==bytesToRead) break;
-                bytesToRead-=dataBytesRead;
-                b+=dataBytesRead;
+            const size_t leftover = LEN_PREFIX_BUFLEN - bytes_read;
+            const ByteBuf_t b = m_rbuf + LEN_PREFIX_BUFLEN;
+            const size_t bytesToRead=length - leftover;
+            doReadFromSocket(b, bytesToRead, error);
+
+            // Check if any error happen while reading the message bytes. If yes then return before decoding the Msg
+            if(error) {
+                DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleHandshake: ERR_CONN_RDFAIL. "
+                                                  << " Failed to read entire handshake message. with error: "
+                                                  << error.message().c_str() << "\n";)
+                handleConnError(CONN_FAILURE, getMessage(ERR_CONN_RDFAIL, "Failed to read entire handshake message"));
+                return;
             }
+
+            // Decode the bytes into a valid RPC Message
             if (!decode(m_rbuf+bytes_read, length, msg)) {
                 DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleHandshake: ERR_CONN_RDFAIL. Cannot decode handshake.\n";)
                 handleConnError(CONN_FAILURE, getMessage(ERR_CONN_RDFAIL, "Cannot decode handshake"));
@@ -340,6 +482,11 @@ void DrillClientImpl::handleHandshake(ByteBuf_t _buf,
             this->m_serverAuthMechanisms.push_back(mechanism);
         }
 
+        // Updated encryption context based on server response
+        this->m_encryptionCtxt.setEncryptionReqd(b2u.has_encrypted() && b2u.encrypted());
+        if(b2u.has_maxwrappedsize()) {
+            this->m_encryptionCtxt.setMaxWrappedSize(b2u.maxwrappedsize());
+        }
     }else{
         // boost error
         if(error==boost::asio::error::eof){ // Server broke off the connection
@@ -360,7 +507,8 @@ void DrillClientImpl::handleHShakeReadTimeout(const boost::system::error_code &
         if (m_deadlineTimer.expires_at() <= boost::asio::deadline_timer::traits_type::now()){
             // The deadline has passed.
             m_deadlineTimer.expires_at(boost::posix_time::pos_infin);
-            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::HandleHShakeReadTimeout: Deadline timer expired; ERR_CONN_HSHAKETIMOUT.\n";)
+            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::HandleHShakeReadTimeout: "
+                                              << "Deadline timer expired; ERR_CONN_HSHAKETIMOUT.\n";)
             handleConnError(CONN_HANDSHAKE_TIMEOUT, getMessage(ERR_CONN_HSHAKETIMOUT));
             m_io_service.stop();
             boost::system::error_code ignorederr;
@@ -370,6 +518,33 @@ void DrillClientImpl::handleHShakeReadTimeout(const boost::system::error_code &
     return;
 }
 
+/*
+ * Check's if client has explicitly expressed interest in encrypted connections only. It looks for USERPROP_SASL_ENCRYPT
+ * connection string property. If set to true then returns true else returns false
+ */
+bool DrillClientImpl::clientNeedsEncryption(const DrillUserProperties* userProperties) {
+    bool needsEncryption = false;
+    // check if userProperties is null
+    if(!userProperties) {
+        return needsEncryption;
+    }
+
+    // Loop through the property to find USERPROP_SASL_ENCRYPT and it's value
+    for (size_t i = 0; i < userProperties->size(); i++) {
+        const std::string key = userProperties->keyAt(i);
+        std::string value = userProperties->valueAt(i);
+
+        if(USERPROP_SASL_ENCRYPT == key) {
+            boost::algorithm::to_lower(value);
+
+            if(0 == value.compare("true")) {
+                needsEncryption = true;
+            }
+        }
+    }
+    return needsEncryption;
+}
+
 connectionStatus_t DrillClientImpl::validateHandshake(DrillUserProperties* properties){
 
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "validateHandShake\n";)
@@ -379,7 +554,7 @@ connectionStatus_t DrillClientImpl::validateHandshake(DrillUserProperties* prope
     u2b.set_rpc_version(DRILL_RPC_VERSION);
     u2b.set_support_listening(true);
     u2b.set_support_timeout(DrillClientConfig::getHeartbeatFrequency() > 0);
-    u2b.set_sasl_support(exec::user::SASL_AUTH);
+    u2b.set_sasl_support(exec::user::SASL_PRIVACY);
 
     // Adding version info
     exec::user::RpcEndpointInfos* infos = u2b.mutable_client_infos();
@@ -436,7 +611,7 @@ connectionStatus_t DrillClientImpl::validateHandshake(DrillUserProperties* prope
         uint64_t coordId = this->getNextCoordinationId();
 
         rpc::OutBoundRpcMessage out_msg(exec::rpc::REQUEST, exec::user::HANDSHAKE, coordId, &u2b);
-        sendSync(out_msg);
+        sendSyncCommon(out_msg);
         DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Sent handshake request message. Coordination id: " << coordId << "\n";)
     }
 
@@ -479,6 +654,13 @@ connectionStatus_t DrillClientImpl::validateHandshake(DrillUserProperties* prope
 }
 
 connectionStatus_t DrillClientImpl::handleAuthentication(const DrillUserProperties *userProperties) {
+
+    // Check if client needs encryption and server is configured for encryption or not before starting handshake
+    if(clientNeedsEncryption(userProperties) && !m_encryptionCtxt.isEncryptionReqd()) {
+        return handleConnError(CONN_AUTH_FAILED, "Client needs encryption but on server side encryption is disabled."
+                                                 " Please check connection parameters or contact administrator?");
+    }
+
     try {
         m_saslAuthenticator = new SaslAuthenticatorImpl(userProperties);
     } catch (std::runtime_error& e) {
@@ -495,26 +677,46 @@ connectionStatus_t DrillClientImpl::handleAuthentication(const DrillUserProperti
         }
     }
 
+    std::stringstream logMsg;
+    logMsg << "DrillClientImpl::handleAuthentication: Authentication failed. [Details: ";
+
     if (SASL_OK == m_saslResultCode) {
-        DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "DrillClientImpl::handleAuthentication: Successfully authenticated!"
-                                          << std::endl;)
+        // Check the negotiated SSF value and change the handlers.
+        if(m_encryptionCtxt.isEncryptionReqd()) {
+            if(SASL_OK != m_saslAuthenticator->verifyAndUpdateSaslProps()) {
+                logMsg << m_encryptionCtxt << "]. Negotiated Parameter is invalid."
+                       << " Error: " << m_saslResultCode;
+                DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << logMsg.str() << std::endl;)
+                return handleConnError(CONN_AUTH_FAILED, logMsg.str().c_str());
+            }
+
+            // Successfully negotiated for encryption related security parameters.
+            // Start using Encrypt and Decrypt handlers.
+            m_fpCurrentSendHandler = &DrillClientImpl::sendSyncEncrypted;
+            m_fpCurrentReadMsgHandler = &DrillClientImpl::readAndDecryptMsg;
+        }
 
-        // in future, negotiated security layers are known here..
+        // Reset the errorMsg stream since this is success case.
+        logMsg.str(std::string());
+        logMsg << "DrillClientImpl::handleAuthentication: Successfully authenticated! [Details: "
+               << m_encryptionCtxt << " ]";
 
+        DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << logMsg.str() << std::endl;)
         m_io_service.reset();
         return CONN_SUCCESS;
     } else {
-        DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "DrillClientImpl::handleAuthentication: Authentication failed: "
-                                          << m_saslResultCode << std::endl;)
+        logMsg << m_encryptionCtxt << ", Error: " << m_saslResultCode;
+        DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << logMsg.str() << std::endl;)
+
         // shuts down socket as well
-        return handleConnError(CONN_AUTH_FAILED, "Authentication failed. Check connection parameters?");
+        logMsg << "]. Check connection parameters?";
+        return handleConnError(CONN_AUTH_FAILED, logMsg.str().c_str());
     }
 }
 
 void DrillClientImpl::initiateAuthentication() {
     exec::shared::SaslMessage response;
-    m_saslResultCode = m_saslAuthenticator->init(m_serverAuthMechanisms, response);
-
+    m_saslResultCode = m_saslAuthenticator->init(m_serverAuthMechanisms, response, &m_encryptionCtxt);
 
     switch (m_saslResultCode) {
         case SASL_CONTINUE:
@@ -539,7 +741,7 @@ void DrillClientImpl::sendSaslResponse(const exec::shared::SaslMessage& response
     boost::lock_guard<boost::mutex> lock(m_dcMutex);
     const int32_t coordId = getNextCoordinationId();
     rpc::OutBoundRpcMessage msg(exec::rpc::REQUEST, exec::user::SASL_MESSAGE, coordId, &response);
-    sendSync(msg);
+    sendSyncCommon(msg);
     if (m_pendingRequests++ == 0) {
         getNextResult();
     }
@@ -768,23 +970,23 @@ Handle* DrillClientImpl::sendMsg(boost::function<Handle*(int32_t)> handleFactory
         phandle = handleFactory(coordId);
         this->m_queryHandles[coordId]=phandle;
 
-        connectionStatus_t cStatus=sendSync(out_msg);
+        connectionStatus_t cStatus = sendSyncCommon(out_msg);
         if(cStatus == CONN_SUCCESS){
             bool sendRequest=false;
 
             DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG)  << "Sent " << ::exec::user::RpcType_Name(type) << " request. " << "[" << m_connectedHost << "]"  << "Coordination id = " << coordId << std::endl;)
-                DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG)  << "Sent " << ::exec::user::RpcType_Name(type) <<  " Coordination id = " << coordId << " query: " << phandle->getQuery() << std::endl;)
+            DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG)  << "Sent " << ::exec::user::RpcType_Name(type) <<  " Coordination id = " << coordId << " query: " << phandle->getQuery() << std::endl;)
 
-                if(m_pendingRequests++==0){
-                    sendRequest=true;
-                }else{
-                    DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "Queuing " << ::exec::user::RpcType_Name(type) <<  " request to server" << std::endl;)
-                        DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "Number of pending requests = " << m_pendingRequests << std::endl;)
-                }
+            if(m_pendingRequests++==0){
+                sendRequest=true;
+            }else{
+                DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "Queuing " << ::exec::user::RpcType_Name(type) <<  " request to server" << std::endl;)
+                DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "Number of pending requests = " << m_pendingRequests << std::endl;)
+            }
             if(sendRequest){
                 DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "Sending " << ::exec::user::RpcType_Name(type) <<  " request. Number of pending requests = "
                         << m_pendingRequests << std::endl;)
-                    getNextResult(); // async wait for results
+                getNextResult(); // async wait for results
             }
         }
 
@@ -854,76 +1056,319 @@ void DrillClientImpl::waitForResults(){
     }
 }
 
-status_t DrillClientImpl::readMsg(ByteBuf_t _buf,
-        AllocatedBufferPtr* allocatedBuffer,
-        rpc::InBoundRpcMessage& msg){
+/*
+ *  Decode the length of the message from bufWithLen and then read entire message from the socket.
+ *  Parameters:
+ *      bufWithLenField            - in  param  - buffer containing the length of the RPC message/encrypted chunk
+ *      bufferWithDataAndLenBytes  - out param  - buffer pointer which points to memory allocated in this function and has the
+ *                                                entire one RPC message / encrypted chunk along with the length of the message.
+ *                                                Memory for this buffer is released by caller.
+ *      lengthFieldLength          - out param  - bytes of bufWithLen which contains the length of the entire RPC message or
+ *                                                encrypted chunk
+ *      lengthDecodeHandler        - in  param  - function pointer with length decoder to use. For encrypted chunk we use
+ *                                                lengthDecode and for plain RPC message we use rpcLengthDecode.
+ *  Return:
+ *      status_t    - QRY_SUCCESS    - In case of success.
+ *                  - QRY_COMM_ERROR/QRY_INTERNAL_ERROR/QRY_CLIENT_OUTOFMEM - In cases of error.
+ */
+status_t DrillClientImpl::readLenBytesFromSocket(const ByteBuf_t bufWithLenField, AllocatedBufferPtr* bufferWithDataAndLenBytes,
+                                                 uint32_t& lengthFieldLength, lengthDecoder lengthDecodeHandler) {
+
+    uint32_t rmsgLen = 0;
+    boost::system::error_code error;
+    *bufferWithDataAndLenBytes = NULL;
+
+    // Decode the length field
+    lengthFieldLength = (this->*lengthDecodeHandler)(bufWithLenField, rmsgLen);
+
+    DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Length bytes = " << lengthFieldLength << std::endl;)
+    DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Msg Length = " << rmsgLen << std::endl;)
+
+    if(rmsgLen>0) {
+        const size_t leftover = LEN_PREFIX_BUFLEN - lengthFieldLength;
+
+        // Allocate a buffer for reading all the bytes in bufWithLen and length number of bytes.
+        const size_t bufferSizeWithLenBytes = rmsgLen + lengthFieldLength;
+        *bufferWithDataAndLenBytes = new AllocatedBuffer(bufferSizeWithLenBytes);
+
+        if(*bufferWithDataAndLenBytes == NULL) {
+            return handleQryError(QRY_CLIENT_OUTOFMEM, getMessage(ERR_QRY_OUTOFMEM), NULL);
+        }
+
+        DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readLenBytesFromSocket: Allocated and locked buffer: [ "
+                                          << *bufferWithDataAndLenBytes << ", size = " << bufferSizeWithLenBytes << " ]\n";)
+
+        // Copy the memory of bufWithLen into bufferWithLenBytesSize
+        memcpy((*bufferWithDataAndLenBytes)->m_pBuffer, bufWithLenField, LEN_PREFIX_BUFLEN);
+        const size_t bytesToRead = rmsgLen - leftover;
+        DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Copied bufWithLen into bufferWithLenBytes. "
+                                          << "Now reading data (rmsgLen - leftover) : " << bytesToRead
+                                          << std::endl;)
+
+        // Read the entire data left from socket and copy to currentBuffer.
+        const ByteBuf_t b = (*bufferWithDataAndLenBytes)->m_pBuffer + LEN_PREFIX_BUFLEN;
+        doReadFromSocket(b, bytesToRead, error);
+    } else {
+        return handleQryError(QRY_INTERNAL_ERROR, getMessage(ERR_QRY_INVREADLEN), NULL);
+    }
+
+    return error ? handleQryError(QRY_COMM_ERROR, getMessage(ERR_QRY_COMMERR, error.message().c_str()), NULL)
+                 : QRY_SUCCESS;
+}
+
+
+/*
+ *  Function to read entire RPC message from socket and decode it to InboundRpcMessage
+ *  Parameters:
+ *      inBuf           - in param  - Buffer containing the length bytes.
+ *      allocatedBuffer - out param - Buffer containing the length bytes and entire RPC message bytes.
+ *      msg             - out param - Decoded InBoundRpcMessage from the bytes in allocatedBuffer
+ *  Return:
+ *      status_t    - QRY_SUCCESS   - In case of success.
+ *                  - QRY_COMM_ERROR/QRY_INTERNAL_ERROR/QRY_CLIENT_OUTOFMEM - In cases of error.
+ */
+status_t DrillClientImpl::readMsg(const ByteBuf_t inBuf, AllocatedBufferPtr* allocatedBuffer,
+                                  rpc::InBoundRpcMessage& msg){
 
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readMsg: Read message from buffer "
-        <<  reinterpret_cast<int*>(_buf) << std::endl;)
-    size_t leftover=0;
-    uint32_t rmsgLen;
-    AllocatedBufferPtr currentBuffer;
-    *allocatedBuffer=NULL;
+                                      <<  reinterpret_cast<int*>(inBuf) << std::endl;)
+    *allocatedBuffer = NULL;
+    {
+        // We need to protect the readLength and read buffer, and the pending requests counter,
+        // but we don't have to keep the lock while we decode the rest of the buffer.
+        boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
+        uint32_t lengthFieldSize = 0;
+
+        // Read the message length and extract length size bytes to form InBoundRpcMessage
+        const status_t statusCode = readLenBytesFromSocket(inBuf, allocatedBuffer, lengthFieldSize,
+                                                           &DrillClientImpl::rpcLengthDecode);
+
+        // Check for error conditions
+        if(QRY_SUCCESS != statusCode) {
+            Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+            return statusCode;
+        }
+
+        // Get the message size
+        size_t msgLen = (*allocatedBuffer)->m_bufSize;
+
+        // Read data successfully, now let's try to decode the buffer and form a valid RPC message.
+        // allocatedBuffer also contains the length bytes which is not needed by decodes so skip that part of buffer.
+        // We have it since in case of encryption the unwrap function expects it
+        if (!decode((*allocatedBuffer)->m_pBuffer + lengthFieldSize, msgLen - lengthFieldSize, msg)) {
+            Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+            return handleQryError(QRY_COMM_ERROR, getMessage(ERR_QRY_COMMERR, "Cannot decode server message"), NULL);
+        }
+
+        DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Successfully created a RPC message with Coordination id: "
+                                          << msg.m_coord_id << std::endl;)
+    }
+    DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readMsg: Free buffer "
+                                      <<  reinterpret_cast<int*>(inBuf) << std::endl;)
+    Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+    return QRY_SUCCESS;
+}
+
+
+/*
+ *  Read ENCRYPT_LEN_PREFIX_BUFLEN bytes to decode length of one complete encrypted chunk. The length bytes are expected
+ *  to be in network order. It is converted to host order and the value is stored in rmsgLen parameter.
+ *  Parameters:
+ *      inBuf   - in param  - ByteBuf_t containing atleast the length bytes.
+ *      rmsgLen - out param - Contain the decoded value of length.
+ *  Return:
+ *      size_t  - length bytes read to decode
+ */
+size_t DrillClientImpl::lengthDecode(const ByteBuf_t inBuf, uint32_t& rmsgLen) {
+    memcpy(&rmsgLen, inBuf, ENCRYPT_LEN_PREFIX_BUFLEN);
+    rmsgLen = ntohl(rmsgLen);
+    return ENCRYPT_LEN_PREFIX_BUFLEN;
+}
+
+/*
+ *  Wrapper which uses RPC message length decoder to get length of one complete RPC message from _buf.
+ *  Parameters:
+ *      inBuf   - in param  - ByteBuf_t containing atleast the length bytes.
+ *      rmsgLen - out param - Contain the decoded value of length.
+ *  Return:
+ *      size_t	- length bytes read to decode
+ */
+size_t DrillClientImpl::rpcLengthDecode(const ByteBuf_t inBuf, uint32_t& rmsgLen) {
+    return rpc::lengthDecode(inBuf, rmsgLen);
+}
+
+
+/*
+ *  Read all the encrypted chunk needed to form a complete RPC message. Read an entire chunk from network, decrypt it
+ *  and put in a buffer. The same process is repeated until the entire buffer to form a completed RPC message is read.
+ *  Parameters:
+ *      inBuf           - in param  - ByteBuf_t containing atleast the length bytes.
+ *      allocatedBuffer - out param - Buffer containing the entire RPC message bytes which is formed by reading all the
+ *                                    required encrypted chunk from network and decrypting each individual chunk. The
+ *                                    buffer memory is released by caller.
+.*      msg             - out param - InBoundRpcMessage formed from bytes in allocatedBuffer
+ *  Return:
+ *      status_t    - QRY_SUCCESS - In case of success.
+ *                  - QRY_COMM_ERROR/QRY_INTERNAL_ERROR/QRY_CLIENT_OUTOFMEM - In cases of error.
+ */
+status_t DrillClientImpl::readAndDecryptMsg(const ByteBuf_t inBuf, AllocatedBufferPtr* allocatedBuffer,
+                                            rpc::InBoundRpcMessage& msg) {
+
+    DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg: Read message from buffer "
+                                      << reinterpret_cast<int*>(inBuf) << std::endl;)
+
+    size_t leftover = 0;
+    uint32_t rpcMsgLen = 0;
+    size_t bytes_read = 0;
+    uint32_t writeIndex = 0;
+    size_t bytesToRead = 0;
+
+    *allocatedBuffer = NULL;
+    boost::system::error_code error;
+    std::stringstream errorMsg;
+
     {
         // We need to protect the readLength and read buffer, and the pending requests counter,
         // but we don't have to keep the lock while we decode the rest of the buffer.
         boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
-        std::size_t bytes_read = rpc::lengthDecode(_buf, rmsgLen);
-        DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "len bytes = " << bytes_read << std::endl;)
-        DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "rmsgLen = " << rmsgLen << std::endl;)
-
-        if(rmsgLen>0){
-            leftover = LEN_PREFIX_BUFLEN - bytes_read;
-            // Allocate a buffer
-            currentBuffer=new AllocatedBuffer(rmsgLen);
-            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readMsg: Allocated and locked buffer: [ "
-                << currentBuffer << ", size = " << rmsgLen << " ]\n";)
-            if(currentBuffer==NULL){
-                Utils::freeBuffer(_buf, LEN_PREFIX_BUFLEN);
-                return handleQryError(QRY_CLIENT_OUTOFMEM, getMessage(ERR_QRY_OUTOFMEM), NULL);
+
+        do{
+            AllocatedBufferPtr currentBuffer = NULL;
+            uint32_t lengthFieldSize = 0;
+            const status_t statusCode = readLenBytesFromSocket(inBuf, &currentBuffer, lengthFieldSize,
+                                                               &DrillClientImpl::lengthDecode);
+
+            if(QRY_SUCCESS != statusCode) {
+                Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+
+                // Release the buffer allocated to hold chunk
+                if(currentBuffer != NULL) {
+                    Utils::freeBuffer(currentBuffer->m_pBuffer, currentBuffer->m_bufSize);
+                    currentBuffer = NULL;
+                }
+                return statusCode;
             }
-            *allocatedBuffer=currentBuffer;
-            if(leftover){
-                memcpy(currentBuffer->m_pBuffer, _buf + bytes_read, leftover);
+
+            // read one chunk successfully. Let's try to decrypt the message
+            const char* unWrappedData = NULL;
+            uint32_t unWrappedLen = 0;
+            const int decryptResult = m_saslAuthenticator->unwrap(reinterpret_cast<const char*>(currentBuffer->m_pBuffer),
+                                                                  currentBuffer->m_bufSize, &unWrappedData, unWrappedLen);
+
+            if(SASL_OK != decryptResult) {
+
+                errorMsg << "Sasl unwrap failed for the buffer of size:" << currentBuffer->m_bufSize << " , Error: "
+                         << decryptResult;
+
+                DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "DrillClientImpl::readAndDecryptMsg: "
+                                                  << errorMsg.str() << std::endl;)
+
+                Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+
+                // Release the buffer allocated to hold chunk
+                Utils::freeBuffer(currentBuffer->m_pBuffer, currentBuffer->m_bufSize);
+                currentBuffer = NULL;
+                return handleQryError(QRY_COMM_ERROR,
+                                      getMessage(ERR_QRY_COMMERR, errorMsg.str().c_str()), NULL);
             }
-            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "reading data (rmsgLen - leftover) : "
-                << (rmsgLen - leftover) << std::endl;)
-            ByteBuf_t b=currentBuffer->m_pBuffer + leftover;
-            size_t bytesToRead=rmsgLen - leftover;
-            boost::system::error_code error;
-            while(1){
-                size_t dataBytesRead=this->m_socket.read_some(
-                        boost::asio::buffer(b, bytesToRead),
-                        error);
-                if(error) break;
-                DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Data Message: actual bytes read = " << dataBytesRead << std::endl;)
-                if(dataBytesRead==bytesToRead) break;
-                bytesToRead-=dataBytesRead;
-                b+=dataBytesRead;
+
+            // Check for case if the unWrappedLen is 0, since Cyrus SASL plugin verifies if the length of wrapped data
+            // is less than the length specified by prepended 4 octets as per RFC 4422/2222. If so it just returns
+            // and waits for more data
+            if(unWrappedLen == 0 || (unWrappedData == NULL)) {
+                errorMsg << "Sasl unwrap failed with mismatch in length of wrapped data and the prepended length value";
+                DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "DrillClientImpl::readAndDecryptMsg: " << errorMsg.str()
+                                                  << std::endl;)
+
+                Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+
+                // Release the buffer allocated to hold chunk
+                Utils::freeBuffer(currentBuffer->m_pBuffer, currentBuffer->m_bufSize);
+                currentBuffer = NULL;
+                return handleQryError(QRY_COMM_ERROR,
+                                      getMessage(ERR_QRY_COMMERR, errorMsg.str().c_str()), NULL);
             }
 
-            if(!error){
-                // read data successfully
-                if (!decode(currentBuffer->m_pBuffer, rmsgLen, msg)) {
-                    Utils::freeBuffer(_buf, LEN_PREFIX_BUFLEN);
-                    return handleQryError(QRY_COMM_ERROR,
-                            getMessage(ERR_QRY_COMMERR, "Cannot decode server message"), NULL);;
+            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg: Successfully decrypted the buffer"
+                                              << " Sizes - Before Decryption =  " << currentBuffer->m_bufSize
+                                              << " and After Decryption = " << unWrappedLen << std::endl;)
+
+            // Release the buffer allocated to hold chunk
+            Utils::freeBuffer(currentBuffer->m_pBuffer, currentBuffer->m_bufSize);
+            currentBuffer = NULL;
+
+            bytes_read = 0;
+            if(*allocatedBuffer == NULL) {
+                // This is the first chunk of the RPC message. We will decode the RPC message full length
+                bytes_read = rpcLengthDecode(reinterpret_cast<ByteBuf_t>(const_cast<char*>(unWrappedData)), rpcMsgLen);
+
+                DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg: Rpc Message Length bytes = "
+                                                  << bytes_read << std::endl;)
+                DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg: Rpc Message Length = "
+                                                  << rpcMsgLen << std::endl;)
+
+                if(rpcMsgLen == 0) {
+                    Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+                    return handleQryError(QRY_INTERNAL_ERROR, getMessage(ERR_QRY_INVREADLEN), NULL);
                 }
-                DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Done decoding chunk. Coordination id: " <<msg.m_coord_id<< std::endl;)
-            }else{
-                Utils::freeBuffer(_buf, LEN_PREFIX_BUFLEN);
-                return handleQryError(QRY_COMM_ERROR,
-                        getMessage(ERR_QRY_COMMERR, error.message().c_str()), NULL);
+                // Allocate a buffer for storing full RPC message. This is released by the caller
+                *allocatedBuffer = new AllocatedBuffer(rpcMsgLen);
+
+                if(*allocatedBuffer == NULL){
+                    Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+                    return handleQryError(QRY_CLIENT_OUTOFMEM, getMessage(ERR_QRY_OUTOFMEM), NULL);
+                }
+
+                DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg:  Allocated and locked buffer:"
+                                                  << "[ " << *allocatedBuffer << ", size = " << rpcMsgLen << " ]\n";)
+
+                bytesToRead = rpcMsgLen;
             }
-        }else{
-            // got a message with an invalid read length.
-            Utils::freeBuffer(_buf, LEN_PREFIX_BUFLEN);
-            return handleQryError(QRY_INTERNAL_ERROR, getMessage(ERR_QRY_INVREADLEN), NULL);
+
+            // Update the leftover bytes that is not copied yet
+            leftover = unWrappedLen - bytes_read;
+
+            // Copy rest of decrypted message to the buffer. We can do this since it is assured that one
+            // entire decrypted chunk is part of the same RPC message.
+            if(leftover) {
+                memcpy((*allocatedBuffer)->m_pBuffer + writeIndex, unWrappedData + bytes_read, leftover);
+            }
+
+            // Update bytes left to read to form full RPC message.
+            bytesToRead -= leftover;
+            writeIndex += leftover;
+
+            DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg: Left to read unencrypted data"
+                                              << " of length (bytesToRead) : " << bytesToRead << std::endl;)
+
+            if(bytesToRead > 0) {
+                // Read synchronously buffer of size LEN_PREFIX_BUFLEN to get length of next chunk
+                doReadFromSocket(inBuf, LEN_PREFIX_BUFLEN, error);
+
+                if(error) {
+                    Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+                    return handleQryError(QRY_COMM_ERROR, getMessage(ERR_QRY_COMMERR, error.message().c_str()), NULL);
+                }
+            }
+        }while(bytesToRead > 0); // more chunks to read for entire RPC message
+
+        DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg: Done decrypting entire RPC message "
+                                          << " of length: " << rpcMsgLen << ". Now starting decode:" << std::endl;)
+
+        // Decode the buffer and form a RPC message
+        if (!decode((*allocatedBuffer)->m_pBuffer, rpcMsgLen, msg)) {
+            Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
+            return handleQryError(QRY_COMM_ERROR, getMessage(ERR_QRY_COMMERR,
+                                  "Cannot decode server message into valid RPC message"), NULL);
         }
+
+        DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Successfully created a RPC message with Coordination id: "
+                                          << msg.m_coord_id << std::endl;)
     }
-    DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readMsg: Free buffer "
-        <<  reinterpret_cast<int*>(_buf) << std::endl;)
-    Utils::freeBuffer(_buf, LEN_PREFIX_BUFLEN);
+
+    DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::readAndDecryptMsg: Free buffer "
+                                      <<  reinterpret_cast<int*>(inBuf) << std::endl;)
+    Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
     return QRY_SUCCESS;
 }
 
@@ -1364,15 +1809,15 @@ status_t DrillClientImpl::processServerMetaResult(AllocatedBufferPtr allocatedBu
     std::map<int,DrillClientQueryHandle*>::const_iterator it=this->m_queryHandles.find(msg.m_coord_id);
     if(it!=this->m_queryHandles.end()){
         DrillClientServerMetaHandle* pHandle=static_cast<DrillClientServerMetaHandle*>((*it).second);
+        exec::user::GetServerMetaResp* resp = new exec::user::GetServerMetaResp();
         DRILL_MT_LOG(DRILL_LOG(LOG_TRACE)  << "Received GetServerMetaResp result Handle " << msg.m_pbody.size() << std::endl;)
-        exec::user::GetServerMetaResp resp;
-        if (!(resp.ParseFromArray(msg.m_pbody.data(), msg.m_pbody.size()))) {
+        if (!(resp->ParseFromArray(msg.m_pbody.data(), msg.m_pbody.size()))) {
             return handleQryError(QRY_COMM_ERROR, "Cannot decode GetServerMetaResp results", pHandle);
         }
-        if (resp.status() != exec::user::OK) {
-            return handleQryError(QRY_FAILED, resp.error(), pHandle);
+        if (resp->status() != exec::user::OK) {
+            return handleQryError(QRY_FAILED, resp->error(), pHandle);
         }
-        pHandle->notifyListener(&(resp.server_meta()), NULL);
+        pHandle->notifyListener(&(resp->server_meta()), NULL);
         DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "GetServerMetaResp result " << std::endl;)
     }else{
         return handleQryError(QRY_INTERNAL_ERROR, getMessage(ERR_QRY_INVQUERYID), NULL);
@@ -1484,11 +1929,11 @@ void DrillClientImpl::handleReadTimeout(const boost::system::error_code & err){
     return;
 }
 
-void DrillClientImpl::handleRead(ByteBuf_t _buf,
+void DrillClientImpl::handleRead(ByteBuf_t inBuf,
         const boost::system::error_code& error,
         size_t bytes_transferred) {
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleRead: Handle Read from buffer "
-        <<  reinterpret_cast<int*>(_buf) << std::endl;)
+        <<  reinterpret_cast<int*>(inBuf) << std::endl;)
     if(DrillClientConfig::getQueryTimeout() > 0){
         // Cancel the timeout if handleRead is called
         DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleRead: Cancel deadline timer.\n";)
@@ -1496,7 +1941,7 @@ void DrillClientImpl::handleRead(ByteBuf_t _buf,
     }
     if (error) {
         // boost error
-        Utils::freeBuffer(_buf, LEN_PREFIX_BUFLEN);
+        Utils::freeBuffer(inBuf, LEN_PREFIX_BUFLEN);
         boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
         DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "DrillClientImpl::handleRead: ERR_QRY_COMMERR. "
             "Boost Communication Error: " << error.message() << std::endl;)
@@ -1510,7 +1955,7 @@ void DrillClientImpl::handleRead(ByteBuf_t _buf,
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Getting new message" << std::endl;)
     AllocatedBufferPtr allocatedBuffer=NULL;
 
-    if(readMsg(_buf, &allocatedBuffer, msg)!=QRY_SUCCESS){
+    if((this->*m_fpCurrentReadMsgHandler)(inBuf, &allocatedBuffer, msg)!=QRY_SUCCESS){
         delete allocatedBuffer;
         if(m_pendingRequests!=0){
             boost::lock_guard<boost::mutex> lock(this->m_dcMutex);
@@ -1655,6 +2100,9 @@ status_t DrillClientImpl::validateResultMessage(const rpc::InBoundRpcMessage& ms
     return QRY_SUCCESS;
 }
 
+/*
+ * Called when there is failure in connect/send.
+ */
 connectionStatus_t DrillClientImpl::handleConnError(connectionStatus_t status, const std::string& msg){
     DrillClientError* pErr = new DrillClientError(status, DrillClientError::CONN_ERROR_START+status, msg);
     m_pendingRequests=0;
@@ -1669,19 +2117,28 @@ connectionStatus_t DrillClientImpl::handleConnError(connectionStatus_t status, c
     return status;
 }
 
+/*
+ * Always called with NULL QueryHandle when there is any error while reading data from socket. Once enough data is read
+ * and a valid RPC message is formed then it can get called with NULL/valid QueryHandle depending on if QueryHandle is found
+ * for the created RPC message.
+ */
 status_t DrillClientImpl::handleQryError(status_t status, const std::string& msg, DrillClientQueryHandle* pQueryHandle){
     DrillClientError* pErr = new DrillClientError(status, DrillClientError::QRY_ERROR_START+status, msg);
-    // set query error only if queries are running
+    // Set query error only if queries are running. If valid QueryHandle that means the bytes to form a valid
+    // RPC message was read successfully from socket. So there is no socket/connection issues.
     if(pQueryHandle!=NULL){
         m_pendingRequests--;
         pQueryHandle->signalError(pErr);
-    }else{
+    }else{ // This means error was while reading from socket, hence call broadcastError which eventually closes socket.
         m_pendingRequests=0;
         broadcastError(pErr);
     }
     return status;
 }
 
+/*
+ * Always called with valid QueryHandle when there is any error processing Query related data.
+ */
 status_t DrillClientImpl::handleQryError(status_t status,
         const exec::shared::DrillPBError& e,
         DrillClientQueryHandle* pQueryHandle){
@@ -1766,7 +2223,7 @@ void DrillClientImpl::sendAck(const rpc::InBoundRpcMessage& msg, bool isOk){
     ack.set_ok(isOk);
     rpc::OutBoundRpcMessage ack_msg(exec::rpc::RESPONSE, exec::user::ACK, msg.m_coord_id, &ack);
     boost::lock_guard<boost::mutex> lock(m_dcMutex);
-    sendSync(ack_msg);
+    sendSyncCommon(ack_msg);
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "ACK sent" << std::endl;)
 }
 
@@ -1774,7 +2231,7 @@ void DrillClientImpl::sendCancel(const exec::shared::QueryId* pQueryId){
     boost::lock_guard<boost::mutex> lock(m_dcMutex);
     uint64_t coordId = this->getNextCoordinationId();
     rpc::OutBoundRpcMessage cancel_msg(exec::rpc::REQUEST, exec::user::CANCEL_QUERY, coordId, pQueryId);
-    sendSync(cancel_msg);
+    sendSyncCommon(cancel_msg);
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "CANCEL sent" << std::endl;)
 }
 
@@ -1783,6 +2240,21 @@ void DrillClientImpl::shutdownSocket(){
     boost::system::error_code ignorederr;
     m_socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignorederr);
     m_bIsConnected=false;
+
+    // Delete the saslAuthenticatorImpl instance since connection is broken. It will recreated on next
+    // call to connect.
+    if(m_saslAuthenticator != NULL) {
+        delete m_saslAuthenticator;
+        m_saslAuthenticator = NULL;
+    }
+
+    // Reset the SASL states.
+    m_saslDone = false;
+    m_saslResultCode = SASL_OK;
+
+    // Reset the encryption context since connection is invalid
+    m_encryptionCtxt.reset();
+
     DRILL_MT_LOG(DRILL_LOG(LOG_TRACE) << "Socket shutdown" << std::endl;)
 }
 
@@ -1799,8 +2271,6 @@ struct ServerMetaContext {
 	boost::mutex m_mutex;
 	boost::condition_variable m_cv;
 
-    ServerMetaContext(): m_done(false), m_status(QRY_SUCCESS), m_serverMeta(), m_mutex(), m_cv() {};
-
 	static status_t listener(void* ctx, const exec::user::ServerMeta* serverMeta, DrillClientError* err) {
 		ServerMetaContext* context = static_cast<ServerMetaContext*>(ctx);
 			if (err) {
diff --git a/contrib/native/client/src/clientlib/drillClientImpl.hpp b/contrib/native/client/src/clientlib/drillClientImpl.hpp
index d37076e6a..852233f8b 100644
--- a/contrib/native/client/src/clientlib/drillClientImpl.hpp
+++ b/contrib/native/client/src/clientlib/drillClientImpl.hpp
@@ -368,16 +368,17 @@ class DrillClientColumnResult: public DrillClientMetadataResult<Metadata::pfnCol
     	DrillClientMetadataResult<Metadata::pfnColumnMetadataListener, meta::ColumnMetadata, meta::DrillColumnMetadata, exec::user::GetColumnsResp>(client, coordId, "getColumns", listener, listenerCtx) {}
 };
 
+// Length Decoder Function Pointer definition
+typedef size_t (DrillClientImpl::*lengthDecoder)(const ByteBuf_t, uint32_t&);
 
 class DrillClientImpl : public DrillClientImplBase{
     public:
         DrillClientImpl():
-            m_coordinationId(1),
             m_handshakeVersion(0),
             m_handshakeStatus(exec::user::SUCCESS),
             m_bIsConnected(false),
             m_saslAuthenticator(NULL),
-    		m_saslResultCode(SASL_OK),
+            m_saslResultCode(SASL_OK),
             m_saslDone(false),
             m_pendingRequests(0),
             m_pError(NULL),
@@ -388,9 +389,11 @@ class DrillClientImpl : public DrillClientImplBase{
             m_heartbeatTimer(m_io_service),
             m_rbuf(NULL),
             m_wbuf(MAX_SOCK_RD_BUFSIZE),
-			m_bIsDirectConnection(false)
+            m_bIsDirectConnection(false)
     {
         m_coordinationId=rand()%1729+1;
+        m_fpCurrentReadMsgHandler = &DrillClientImpl::readMsg;
+        m_fpCurrentSendHandler = &DrillClientImpl::sendSyncPlain;
     };
 
         ~DrillClientImpl(){
@@ -477,9 +480,10 @@ class DrillClientImpl : public DrillClientImplBase{
         void handleHeartbeatTimeout(const boost::system::error_code & err); // send a heartbeat. If send fails, broadcast error, close connection and bail out.
 
         int32_t getNextCoordinationId(){ return ++m_coordinationId; };
-        // send synchronous messages
-        //connectionStatus_t recvSync(rpc::InBoundRpcMessage& msg);
-        connectionStatus_t sendSync(rpc::OutBoundRpcMessage& msg);
+        // synchronous message send handlers
+        connectionStatus_t sendSyncCommon(rpc::OutBoundRpcMessage& msg);
+        connectionStatus_t sendSyncPlain();
+        connectionStatus_t sendSyncEncrypted();
         // handshake
         connectionStatus_t recvHandshake();
         void handleHandshake(ByteBuf_t b, const boost::system::error_code& err, std::size_t bytes_transferred );
@@ -488,10 +492,16 @@ class DrillClientImpl : public DrillClientImplBase{
         void startMessageListener(); 
         // Query results
         void getNextResult();
-        status_t readMsg(
-                ByteBuf_t _buf,
-                AllocatedBufferPtr* allocatedBuffer,
-                rpc::InBoundRpcMessage& msg);
+        // Read Message Handlers
+        status_t readMsg(const ByteBuf_t inBuf, AllocatedBufferPtr* allocatedBuffer, rpc::InBoundRpcMessage& msg);
+        status_t readAndDecryptMsg(const ByteBuf_t inBuf, AllocatedBufferPtr* allocatedBuffer, rpc::InBoundRpcMessage& msg);
+        status_t readLenBytesFromSocket(const ByteBuf_t bufWithLenField, AllocatedBufferPtr* bufferWithDataAndLenBytes,
+                                        uint32_t& lengthFieldLength, lengthDecoder lengthDecodeHandler);
+        void doReadFromSocket(ByteBuf_t inBuf, size_t bytesToRead, boost::system::error_code& errorCode);
+        void doWriteToSocket(const char* dataPtr, size_t bytesToWrite, boost::system::error_code& errorCode);
+        // Length decode handlers
+        size_t lengthDecode(const ByteBuf_t inBuf, uint32_t& rmsgLen);
+        size_t rpcLengthDecode(const ByteBuf_t inBuf, uint32_t& rmsgLen);
         status_t processQueryResult(AllocatedBufferPtr allocatedBuffer, const rpc::InBoundRpcMessage& msg);
         status_t processQueryData(AllocatedBufferPtr allocatedBuffer, const rpc::InBoundRpcMessage& msg);
         status_t processCancelledQueryResult( exec::shared::QueryId& qid, exec::shared::QueryResult* qr);
@@ -506,7 +516,7 @@ class DrillClientImpl : public DrillClientImplBase{
         status_t processQueryStatusResult( exec::shared::QueryResult* qr,
                 DrillClientQueryResult* pDrillClientQueryResult);
         void handleReadTimeout(const boost::system::error_code & err);
-        void handleRead(ByteBuf_t _buf, const boost::system::error_code & err, size_t bytes_transferred) ;
+        void handleRead(ByteBuf_t inBuf, const boost::system::error_code & err, size_t bytes_transferred) ;
         status_t validateDataMessage(const rpc::InBoundRpcMessage& msg, const exec::shared::QueryData& qd, std::string& valError);
         status_t validateResultMessage(const rpc::InBoundRpcMessage& msg, const exec::shared::QueryResult& qr, std::string& valError);
         connectionStatus_t handleConnError(connectionStatus_t status, const std::string& msg);
@@ -540,6 +550,7 @@ class DrillClientImpl : public DrillClientImplBase{
         void finishAuthentication();
 
         void shutdownSocket();
+        bool clientNeedsEncryption(const DrillUserProperties* userProperties);
 
         int32_t m_coordinationId;
         int32_t m_handshakeVersion;
@@ -557,6 +568,14 @@ class DrillClientImpl : public DrillClientImplBase{
         boost::mutex m_saslMutex; // mutex to protect m_saslDone
         boost::condition_variable m_saslCv; // to signal completion of SASL exchange
 
+        // Used for encryption and is set when server notifies in first handshake response.
+        EncryptionContext m_encryptionCtxt;
+
+        // Function pointer for read and send handler. By default these are referred to handler for plain message read/send. When encryption is enabled
+        // then after successful handshake these pointers refer to handler for encrypted message read/send over wire.
+        status_t (DrillClientImpl::*m_fpCurrentReadMsgHandler)(ByteBuf_t inBuf, AllocatedBufferPtr* allocatedBuffer, rpc::InBoundRpcMessage& msg);
+        connectionStatus_t (DrillClientImpl::*m_fpCurrentSendHandler)();
+
         std::string m_connectStr; 
 
         // 
diff --git a/contrib/native/client/src/clientlib/rpcMessage.cpp b/contrib/native/client/src/clientlib/rpcMessage.cpp
index 13cd7a82e..f64167f5a 100644
--- a/contrib/native/client/src/clientlib/rpcMessage.cpp
+++ b/contrib/native/client/src/clientlib/rpcMessage.cpp
@@ -47,7 +47,7 @@ std::size_t lengthDecode(const uint8_t* buf, uint32_t& length) {
 
     // read the frame to get the length of the message and then
 
-    CodedInputStream cis(buf, 5); // read 5 bytes at most
+    CodedInputStream cis(buf, LEN_PREFIX_BUFLEN); // read LEN_PREFIX_BUFLEN bytes at most
 
     int startPos(cis.CurrentPosition()); // for debugging
     if (!cis.ReadVarint32(&length)) {
diff --git a/contrib/native/client/src/clientlib/rpcMessage.hpp b/contrib/native/client/src/clientlib/rpcMessage.hpp
index 15487e9ba..43bcaeb13 100644
--- a/contrib/native/client/src/clientlib/rpcMessage.hpp
+++ b/contrib/native/client/src/clientlib/rpcMessage.hpp
@@ -54,7 +54,6 @@ std::size_t lengthDecode(const uint8_t* buf, uint32_t& length);
 bool decode(const uint8_t* buf, int length, InBoundRpcMessage& msg);
 
 bool encode(DataBuf& buf, const OutBoundRpcMessage& msg);
-
 } // namespace rpc
 } // namespace Drill
 
diff --git a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp
index e7e2ba594..c5dc3aced 100644
--- a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp
+++ b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.cpp
@@ -32,6 +32,7 @@ static const std::string DEFAULT_SERVICE_NAME = "drill";
 static const std::string KERBEROS_SIMPLE_NAME = "kerberos";
 static const std::string KERBEROS_SASL_NAME = "gssapi";
 static const std::string PLAIN_NAME = "plain";
+static const int PREFERRED_MIN_SSF = 56;
 
 const std::map<std::string, std::string> SaslAuthenticatorImpl::MECHANISM_MAPPING = boost::assign::map_list_of
     (KERBEROS_SIMPLE_NAME, KERBEROS_SASL_NAME)
@@ -42,8 +43,7 @@ boost::mutex SaslAuthenticatorImpl::s_mutex;
 bool SaslAuthenticatorImpl::s_initialized = false;
 
 SaslAuthenticatorImpl::SaslAuthenticatorImpl(const DrillUserProperties* const properties) :
-    m_pUserProperties(properties), m_pConnection(NULL), m_ppwdSecret(NULL) {
-
+    m_pUserProperties(properties), m_pConnection(NULL), m_ppwdSecret(NULL), m_pEncryptCtxt(NULL) {
     if (!s_initialized) {
         boost::lock_guard<boost::mutex> lock(SaslAuthenticatorImpl::s_mutex);
         if (!s_initialized) {
@@ -85,6 +85,9 @@ SaslAuthenticatorImpl::~SaslAuthenticatorImpl() {
         sasl_dispose(&m_pConnection);
     }
     m_pConnection = NULL;
+
+    // Memory is owned by DrillClientImpl object
+    m_pEncryptCtxt = NULL;
 }
 
 typedef int (*sasl_callback_proc_t)(void); // see sasl_callback_ft
@@ -109,8 +112,14 @@ int SaslAuthenticatorImpl::passwordCallback(sasl_conn_t *conn, void *context, in
     return SASL_OK;
 }
 
-int SaslAuthenticatorImpl::init(const std::vector<std::string>& mechanisms, exec::shared::SaslMessage& response) {
-    // find and set parameters
+int SaslAuthenticatorImpl::init(const std::vector<std::string>& mechanisms, exec::shared::SaslMessage& response,
+                                EncryptionContext* const encryptCtxt) {
+
+    // EncryptionContext should not be NULL here.
+    assert(encryptCtxt != NULL);
+    m_pEncryptCtxt = encryptCtxt;
+
+	// find and set parameters
     std::string authMechanismToUse;
     std::string serviceName;
     std::string serviceHost;
@@ -163,6 +172,9 @@ int SaslAuthenticatorImpl::init(const std::vector<std::string>& mechanisms, exec
                                       << saslResult << std::endl;)
     if (saslResult != SASL_OK) return saslResult;
 
+    // set the security properties
+    setSecurityProps();
+
     // initiate; for now, pass in only one mechanism
     const char *out;
     unsigned outlen;
@@ -204,4 +216,103 @@ int SaslAuthenticatorImpl::step(const exec::shared::SaslMessage& challenge, exec
     return saslResult;
 }
 
+/*
+ * Verify that the negotiated value is correct as per system configurations. Also retrieves and set the rawWrapSendSize
+ */
+int SaslAuthenticatorImpl::verifyAndUpdateSaslProps() {
+    const int* negotiatedValue;
+    int result = SASL_OK;
+
+    if(SASL_OK != (result = sasl_getprop(m_pConnection, SASL_SSF, reinterpret_cast<const void **>(&negotiatedValue)))) {
+        return result;
+    }
+
+    // If the negotiated SSF value is less than required one that means we have negotiated for weaker security level.
+    if(*negotiatedValue < PREFERRED_MIN_SSF) {
+        DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "SaslAuthenticatorImpl::verifyAndUpdateSaslProps: "
+                                          << "Negotiated SSF parameter:" << *negotiatedValue
+                                          << " is less than Preferred one: " << PREFERRED_MIN_SSF << std::endl;)
+        result = SASL_BADPARAM;
+        return result;
+    }
+
+    if(SASL_OK != (result = sasl_getprop(m_pConnection, SASL_MAXOUTBUF,
+                                         reinterpret_cast<const void **>(&negotiatedValue)))) {
+        return result;
+    }
+
+    DRILL_MT_LOG(DRILL_LOG(LOG_DEBUG) << "SaslAuthenticatorImpl::verifyAndUpdateSaslProps: "
+                                      << "Negotiated Raw Wrap Buffer size: " << *negotiatedValue << std::endl;)
+
+    m_pEncryptCtxt->setWrapSizeLimit(*negotiatedValue);
+    return result;
+}
+
+/*
+ *  Set the security properties structure with all the needed parameters for encryption so that
+ *  a proper mechanism with and cipher is chosen after handshake.
+ *
+ *  PREFERRED_MIN_SSF is chosen to be 56 since that is the max_ssf supported by gssapi. We want 
+ *  stronger cipher algorithm to be used all the time (preferably AES-256), so leaving MAX_SSF as UINT_MAX
+ */
+void SaslAuthenticatorImpl::setSecurityProps() const{
+
+    if(m_pEncryptCtxt->isEncryptionReqd()) {
+        // set the security properties.
+        sasl_security_properties_t secprops;
+        secprops.min_ssf = PREFERRED_MIN_SSF;
+        secprops.max_ssf = UINT_MAX;
+        secprops.maxbufsize = m_pEncryptCtxt->getMaxWrappedSize();
+        secprops.property_names = NULL;
+        secprops.property_values = NULL;
+        // Only specify NOPLAINTEXT for encryption since the mechanism is selected based on name not
+        // the security properties configured here.
+        secprops.security_flags = SASL_SEC_NOPLAINTEXT;
+
+        // Set the security properties in the connection context.
+        sasl_setprop(m_pConnection, SASL_SEC_PROPS, &secprops);
+    }
+}
+
+/*
+ * Encodes the input data by calling the sasl_encode provided by Cyrus-SASL library which internally calls
+ * the wrap function of the chosen mechanism. The output buffer will have first 4 octets as the length of
+ * encrypted data in network byte order.
+ *
+ * Parameters:
+ *      dataToWrap      -   in param    -   pointer to data buffer to encrypt.
+ *      dataToWrapLen   -   in param    -   length of data buffer to encrypt.
+ *      output          -   out param   -   pointer to data buffer with encrypted data. Allocated by Cyrus-SASL
+ *      wrappedLen      -   out param   -   length of data after encryption
+ * Returns:
+ *      SASL_OK         - success (returns input if no layer negotiated)
+ *      SASL_NOTDONE    - security layer negotiation not finished
+ *      SASL_BADPARAM   - inputlen is greater than the SASL_MAXOUTBUF
+ */
+int SaslAuthenticatorImpl::wrap(const char* dataToWrap, const int& dataToWrapLen, const char** output,
+                                uint32_t& wrappedLen) {
+    return sasl_encode(m_pConnection, dataToWrap, dataToWrapLen, output, &wrappedLen);
+}
+
+/*
+ * Decodes the input data by calling the sasl_decode provided by Cyrus-SASL library which internally calls
+ * the wrap function of the chosen mechanism. The input buffer will have first 4 octets as the length of
+ * encrypted data in network byte order.
+ *
+ * Parameters:
+ *      dataToUnWrap      -   in param    -   pointer to data buffer to decrypt.
+ *      dataToUnWrapLen   -   in param    -   length of data buffer to decrypt.
+ *      output            -   out param   -   pointer to data buffer with decrypted data. Allocated by Cyrus-SASL
+ *      unWrappedLen      -   out param   -   length of data after decryption
+ * Returns:
+ *      SASL_OK         - success (returns input if no layer negotiated)
+ *      SASL_NOTDONE    - security layer negotiation not finished
+ *      SASL_BADPARAM   - inputlen is greater than the SASL_MAXOUTBUF
+ */
+int SaslAuthenticatorImpl::unwrap(const char* dataToUnWrap, const int& dataToUnWrapLen, const char** output,
+                                  uint32_t& unWrappedLen) {
+    return sasl_decode(m_pConnection, dataToUnWrap, dataToUnWrapLen, output, &unWrappedLen);
+}
+
+
 } /* namespace Drill */
diff --git a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp
index 5e36ee123..53fe4e362 100644
--- a/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp
+++ b/contrib/native/client/src/clientlib/saslAuthenticatorImpl.hpp
@@ -24,6 +24,7 @@
 #include <vector>
 #include "drill/drillClient.hpp"
 #include "UserBitShared.pb.h"
+#include "utils.hpp"
 
 #include "sasl/sasl.h"
 #include "sasl/saslplug.h"
@@ -38,10 +39,17 @@ public:
 
     ~SaslAuthenticatorImpl();
 
-    int init(const std::vector<std::string>& mechanisms, exec::shared::SaslMessage& response);
+    int init(const std::vector<std::string>& mechanisms, exec::shared::SaslMessage& response,
+             EncryptionContext* const encryptCtxt);
 
     int step(const exec::shared::SaslMessage& challenge, exec::shared::SaslMessage& response) const;
 
+    int verifyAndUpdateSaslProps();
+
+    int wrap(const char* dataToWrap, const int& dataToWrapLen, const char** output, uint32_t& wrappedLen);
+
+    int unwrap(const char* dataToUnWrap, const int& dataToUnWrapLen, const char** output, uint32_t& unWrappedLen);
+
 private:
 
     static const std::map<std::string, std::string> MECHANISM_MAPPING;
@@ -53,11 +61,13 @@ private:
     sasl_conn_t *m_pConnection;
     std::string m_username;
     sasl_secret_t *m_ppwdSecret;
+    EncryptionContext *m_pEncryptCtxt;
 
     static int passwordCallback(sasl_conn_t *conn, void *context, int id, sasl_secret_t **psecret);
 
     static int userNameCallback(void *context, int id, const char **result, unsigned int *len);
 
+    void setSecurityProps() const;
 };
 
 } /* namespace Drill */
diff --git a/contrib/native/client/src/clientlib/utils.cpp b/contrib/native/client/src/clientlib/utils.cpp
index d3c8f08d7..11aa2c272 100644
--- a/contrib/native/client/src/clientlib/utils.cpp
+++ b/contrib/native/client/src/clientlib/utils.cpp
@@ -111,4 +111,53 @@ AllocatedBuffer::~AllocatedBuffer(){
     m_bufSize = 0;
 }
 
+EncryptionContext::EncryptionContext(const bool& encryptionReqd, const int& maxWrappedSize, const int& wrapSizeLimit) {
+    this->m_bEncryptionReqd = encryptionReqd;
+    this->m_maxWrappedSize = maxWrappedSize;
+    this->m_wrapSizeLimit = wrapSizeLimit;
+}
+
+EncryptionContext::EncryptionContext() {
+    this->m_bEncryptionReqd = false;
+    this->m_maxWrappedSize = 65536;
+    this->m_wrapSizeLimit = 0;
+}
+
+void EncryptionContext::setEncryptionReqd(const bool& encryptionReqd) {
+    this->m_bEncryptionReqd = encryptionReqd;
+}
+
+void EncryptionContext::setMaxWrappedSize(const int& maxWrappedSize) {
+    this->m_maxWrappedSize = maxWrappedSize;
+}
+
+void EncryptionContext::setWrapSizeLimit(const int& wrapSizeLimit) {
+    this->m_wrapSizeLimit = wrapSizeLimit;
+}
+
+bool EncryptionContext::isEncryptionReqd() const {
+    return m_bEncryptionReqd;
+}
+
+int EncryptionContext::getMaxWrappedSize() const {
+    return m_maxWrappedSize;
+}
+
+int EncryptionContext::getWrapSizeLimit() const {
+    return m_wrapSizeLimit;
+}
+
+void EncryptionContext::reset() {
+    this->m_bEncryptionReqd = false;
+    this->m_maxWrappedSize = 65536;
+    this->m_wrapSizeLimit = 0;
+}
+
+std::ostream& operator<<(std::ostream &contextStream, const EncryptionContext& context) {
+    contextStream << " Encryption: " << (context.isEncryptionReqd() ? "enabled" : "disabled");
+    contextStream << " ,MaxWrappedSize: " << context.getMaxWrappedSize();
+    contextStream << " ,WrapSizeLimit: " << context.getWrapSizeLimit();
+    return contextStream;
+}
+
 } // namespace 
diff --git a/contrib/native/client/src/clientlib/utils.hpp b/contrib/native/client/src/clientlib/utils.hpp
index 4cd8fa5e8..d30794c9a 100644
--- a/contrib/native/client/src/clientlib/utils.hpp
+++ b/contrib/native/client/src/clientlib/utils.hpp
@@ -98,6 +98,38 @@ class DECLSPEC_DRILL_CLIENT Utils{
 
 }; // Utils
 
+/*
+ * Encryption related configuration parameters. The member's are updated with value received from server
+ * and also after the SASL Handshake is done.
+ */
+class EncryptionContext {
+
+	bool m_bEncryptionReqd;
+	int m_maxWrappedSize;
+	int m_wrapSizeLimit;
+
+public:
+	EncryptionContext();
+
+	EncryptionContext(const bool& encryptionReqd, const int& maxWrappedSize, const int& wrapSizeLimit);
+
+	void setEncryptionReqd(const bool& encryptionReqd);
+
+	void setMaxWrappedSize(const int& maxWrappedSize);
+
+	void setWrapSizeLimit(const int& wrapSizeLimit);
+
+	bool isEncryptionReqd() const;
+
+	int getMaxWrappedSize() const;
+
+	int getWrapSizeLimit() const;
+
+	void reset();
+
+	friend std::ostream& operator<<(std::ostream &contextStream, const EncryptionContext& context);
+};
+
 } // namespace Drill
 
 #endif
diff --git a/contrib/native/client/src/include/drill/common.hpp b/contrib/native/client/src/include/drill/common.hpp
index ed0a1ed32..5401c75a9 100644
--- a/contrib/native/client/src/include/drill/common.hpp
+++ b/contrib/native/client/src/include/drill/common.hpp
@@ -56,6 +56,7 @@
 
 #define LENGTH_PREFIX_MAX_LENGTH 5
 #define LEN_PREFIX_BUFLEN LENGTH_PREFIX_MAX_LENGTH
+#define ENCRYPT_LEN_PREFIX_BUFLEN 4
 
 #define MAX_CONNECT_STR 4096
 #define MAX_SOCK_RD_BUFSIZE  1024
@@ -169,6 +170,7 @@ typedef enum{
 #define USERPROP_AUTH_MECHANISM "auth"
 #define USERPROP_SERVICE_NAME "service_name"
 #define USERPROP_SERVICE_HOST "service_host"
+#define USERPROP_SASL_ENCRYPT "sasl_encrypt"
 
 // Bitflags to describe user properties
 // Used in DrillUserProperties::USER_PROPERTIES
diff --git a/contrib/native/client/src/protobuf/User.pb.cc b/contrib/native/client/src/protobuf/User.pb.cc
index be3f0016b..e2ae3d19e 100644
--- a/contrib/native/client/src/protobuf/User.pb.cc
+++ b/contrib/native/client/src/protobuf/User.pb.cc
@@ -269,7 +269,7 @@ void protobuf_AssignDesc_User_2eproto() {
       ::google::protobuf::MessageFactory::generated_factory(),
       sizeof(QueryPlanFragments));
   BitToUserHandshake_descriptor_ = file->message_type(7);
-  static const int BitToUserHandshake_offsets_[7] = {
+  static const int BitToUserHandshake_offsets_[9] = {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, rpc_version_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, status_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, errorid_),
@@ -277,6 +277,8 @@ void protobuf_AssignDesc_User_2eproto() {
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, server_infos_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, authenticationmechanisms_),
     GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, supported_methods_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, encrypted_),
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(BitToUserHandshake, maxwrappedsize_),
   };
   BitToUserHandshake_reflection_ =
     new ::google::protobuf::internal::GeneratedMessageReflection(
@@ -952,198 +954,200 @@ void protobuf_AddDesc_User_2eproto() {
     "ed.QueryResult.QueryState\022&\n\010query_id\030\002 "
     "\001(\0132\024.exec.shared.QueryId\0221\n\tfragments\030\003"
     " \003(\0132\036.exec.bit.control.PlanFragment\022(\n\005"
-    "error\030\004 \001(\0132\031.exec.shared.DrillPBError\"\200"
+    "error\030\004 \001(\0132\031.exec.shared.DrillPBError\"\253"
     "\002\n\022BitToUserHandshake\022\023\n\013rpc_version\030\002 \001"
     "(\005\022*\n\006status\030\003 \001(\0162\032.exec.user.Handshake"
     "Status\022\017\n\007errorId\030\004 \001(\t\022\024\n\014errorMessage\030"
     "\005 \001(\t\0221\n\014server_infos\030\006 \001(\0132\033.exec.user."
     "RpcEndpointInfos\022 \n\030authenticationMechan"
     "isms\030\007 \003(\t\022-\n\021supported_methods\030\010 \003(\0162\022."
-    "exec.user.RpcType\"-\n\nLikeFilter\022\017\n\007patte"
-    "rn\030\001 \001(\t\022\016\n\006escape\030\002 \001(\t\"D\n\016GetCatalogsR"
-    "eq\0222\n\023catalog_name_filter\030\001 \001(\0132\025.exec.u"
-    "ser.LikeFilter\"M\n\017CatalogMetadata\022\024\n\014cat"
-    "alog_name\030\001 \001(\t\022\023\n\013description\030\002 \001(\t\022\017\n\007"
-    "connect\030\003 \001(\t\"\223\001\n\017GetCatalogsResp\022(\n\006sta"
-    "tus\030\001 \001(\0162\030.exec.user.RequestStatus\022,\n\010c"
-    "atalogs\030\002 \003(\0132\032.exec.user.CatalogMetadat"
-    "a\022(\n\005error\030\003 \001(\0132\031.exec.shared.DrillPBEr"
-    "ror\"v\n\rGetSchemasReq\0222\n\023catalog_name_fil"
-    "ter\030\001 \001(\0132\025.exec.user.LikeFilter\0221\n\022sche"
-    "ma_name_filter\030\002 \001(\0132\025.exec.user.LikeFil"
-    "ter\"i\n\016SchemaMetadata\022\024\n\014catalog_name\030\001 "
-    "\001(\t\022\023\n\013schema_name\030\002 \001(\t\022\r\n\005owner\030\003 \001(\t\022"
-    "\014\n\004type\030\004 \001(\t\022\017\n\007mutable\030\005 \001(\t\"\220\001\n\016GetSc"
-    "hemasResp\022(\n\006status\030\001 \001(\0162\030.exec.user.Re"
-    "questStatus\022*\n\007schemas\030\002 \003(\0132\031.exec.user"
-    ".SchemaMetadata\022(\n\005error\030\003 \001(\0132\031.exec.sh"
-    "ared.DrillPBError\"\302\001\n\014GetTablesReq\0222\n\023ca"
-    "talog_name_filter\030\001 \001(\0132\025.exec.user.Like"
-    "Filter\0221\n\022schema_name_filter\030\002 \001(\0132\025.exe"
-    "c.user.LikeFilter\0220\n\021table_name_filter\030\003"
-    " \001(\0132\025.exec.user.LikeFilter\022\031\n\021table_typ"
-    "e_filter\030\004 \003(\t\"\\\n\rTableMetadata\022\024\n\014catal"
-    "og_name\030\001 \001(\t\022\023\n\013schema_name\030\002 \001(\t\022\022\n\nta"
-    "ble_name\030\003 \001(\t\022\014\n\004type\030\004 \001(\t\"\215\001\n\rGetTabl"
-    "esResp\022(\n\006status\030\001 \001(\0162\030.exec.user.Reque"
-    "stStatus\022(\n\006tables\030\002 \003(\0132\030.exec.user.Tab"
-    "leMetadata\022(\n\005error\030\003 \001(\0132\031.exec.shared."
-    "DrillPBError\"\333\001\n\rGetColumnsReq\0222\n\023catalo"
-    "g_name_filter\030\001 \001(\0132\025.exec.user.LikeFilt"
-    "er\0221\n\022schema_name_filter\030\002 \001(\0132\025.exec.us"
-    "er.LikeFilter\0220\n\021table_name_filter\030\003 \001(\013"
-    "2\025.exec.user.LikeFilter\0221\n\022column_name_f"
-    "ilter\030\004 \001(\0132\025.exec.user.LikeFilter\"\251\003\n\016C"
-    "olumnMetadata\022\024\n\014catalog_name\030\001 \001(\t\022\023\n\013s"
-    "chema_name\030\002 \001(\t\022\022\n\ntable_name\030\003 \001(\t\022\023\n\013"
-    "column_name\030\004 \001(\t\022\030\n\020ordinal_position\030\005 "
-    "\001(\005\022\025\n\rdefault_value\030\006 \001(\t\022\023\n\013is_nullabl"
-    "e\030\007 \001(\010\022\021\n\tdata_type\030\010 \001(\t\022\027\n\017char_max_l"
-    "ength\030\t \001(\005\022\031\n\021char_octet_length\030\n \001(\005\022\031"
-    "\n\021numeric_precision\030\013 \001(\005\022\037\n\027numeric_pre"
-    "cision_radix\030\014 \001(\005\022\025\n\rnumeric_scale\030\r \001("
-    "\005\022\033\n\023date_time_precision\030\016 \001(\005\022\025\n\rinterv"
-    "al_type\030\017 \001(\t\022\032\n\022interval_precision\030\020 \001("
-    "\005\022\023\n\013column_size\030\021 \001(\005\"\220\001\n\016GetColumnsRes"
-    "p\022(\n\006status\030\001 \001(\0162\030.exec.user.RequestSta"
-    "tus\022*\n\007columns\030\002 \003(\0132\031.exec.user.ColumnM"
-    "etadata\022(\n\005error\030\003 \001(\0132\031.exec.shared.Dri"
-    "llPBError\"/\n\032CreatePreparedStatementReq\022"
-    "\021\n\tsql_query\030\001 \001(\t\"\326\003\n\024ResultColumnMetad"
-    "ata\022\024\n\014catalog_name\030\001 \001(\t\022\023\n\013schema_name"
-    "\030\002 \001(\t\022\022\n\ntable_name\030\003 \001(\t\022\023\n\013column_nam"
-    "e\030\004 \001(\t\022\r\n\005label\030\005 \001(\t\022\021\n\tdata_type\030\006 \001("
-    "\t\022\023\n\013is_nullable\030\007 \001(\010\022\021\n\tprecision\030\010 \001("
-    "\005\022\r\n\005scale\030\t \001(\005\022\016\n\006signed\030\n \001(\010\022\024\n\014disp"
-    "lay_size\030\013 \001(\005\022\022\n\nis_aliased\030\014 \001(\010\0225\n\rse"
-    "archability\030\r \001(\0162\036.exec.user.ColumnSear"
-    "chability\0223\n\014updatability\030\016 \001(\0162\035.exec.u"
-    "ser.ColumnUpdatability\022\026\n\016auto_increment"
-    "\030\017 \001(\010\022\030\n\020case_sensitivity\030\020 \001(\010\022\020\n\010sort"
-    "able\030\021 \001(\010\022\022\n\nclass_name\030\022 \001(\t\022\023\n\013is_cur"
-    "rency\030\024 \001(\010\".\n\027PreparedStatementHandle\022\023"
-    "\n\013server_info\030\001 \001(\014\"\200\001\n\021PreparedStatemen"
-    "t\0220\n\007columns\030\001 \003(\0132\037.exec.user.ResultCol"
-    "umnMetadata\0229\n\rserver_handle\030\002 \001(\0132\".exe"
-    "c.user.PreparedStatementHandle\"\253\001\n\033Creat"
-    "ePreparedStatementResp\022(\n\006status\030\001 \001(\0162\030"
-    ".exec.user.RequestStatus\0228\n\022prepared_sta"
-    "tement\030\002 \001(\0132\034.exec.user.PreparedStateme"
-    "nt\022(\n\005error\030\003 \001(\0132\031.exec.shared.DrillPBE"
-    "rror\"\022\n\020GetServerMetaReq\"P\n\016ConvertSuppo"
-    "rt\022\037\n\004from\030\001 \002(\0162\021.common.MinorType\022\035\n\002t"
-    "o\030\002 \002(\0162\021.common.MinorType\"\223\001\n\021GetServer"
-    "MetaResp\022(\n\006status\030\001 \001(\0162\030.exec.user.Req"
-    "uestStatus\022*\n\013server_meta\030\002 \001(\0132\025.exec.u"
-    "ser.ServerMeta\022(\n\005error\030\003 \001(\0132\031.exec.sha"
-    "red.DrillPBError\"\377\r\n\nServerMeta\022\035\n\025all_t"
-    "ables_selectable\030\001 \001(\010\022%\n\035blob_included_"
-    "in_max_row_size\030\002 \001(\010\022\030\n\020catalog_at_star"
-    "t\030\003 \001(\010\022\031\n\021catalog_separator\030\004 \001(\t\022\024\n\014ca"
-    "talog_term\030\005 \001(\t\0222\n\017collate_support\030\006 \003("
-    "\0162\031.exec.user.CollateSupport\022!\n\031column_a"
-    "liasing_supported\030\007 \001(\010\0222\n\017convert_suppo"
-    "rt\030\010 \003(\0132\031.exec.user.ConvertSupport\022E\n\031c"
-    "orrelation_names_support\030\t \001(\0162\".exec.us"
-    "er.CorrelationNamesSupport\022\033\n\023date_time_"
-    "functions\030\n \003(\t\022F\n\032date_time_literals_su"
-    "pport\030\013 \003(\0162\".exec.user.DateTimeLiterals"
-    "Support\0223\n\020group_by_support\030\014 \001(\0162\031.exec"
-    ".user.GroupBySupport\0226\n\021identifier_casin"
-    "g\030\r \001(\0162\033.exec.user.IdentifierCasing\022\037\n\027"
-    "identifier_quote_string\030\016 \001(\t\022$\n\034like_es"
-    "cape_clause_supported\030\017 \001(\010\022!\n\031max_binar"
-    "y_literal_length\030\020 \001(\r\022\037\n\027max_catalog_na"
-    "me_length\030\021 \001(\r\022\037\n\027max_char_literal_leng"
-    "th\030\022 \001(\r\022\036\n\026max_column_name_length\030\023 \001(\r"
-    "\022\037\n\027max_columns_in_group_by\030\024 \001(\r\022\037\n\027max"
-    "_columns_in_order_by\030\025 \001(\r\022\035\n\025max_column"
-    "s_in_select\030\026 \001(\r\022\036\n\026max_cursor_name_len"
-    "gth\030\027 \001(\r\022\034\n\024max_logical_lob_size\030\030 \001(\r\022"
-    "\024\n\014max_row_size\030\031 \001(\r\022\036\n\026max_schema_name"
-    "_length\030\032 \001(\r\022\034\n\024max_statement_length\030\033 "
-    "\001(\r\022\026\n\016max_statements\030\034 \001(\r\022\035\n\025max_table"
-    "_name_length\030\035 \001(\r\022\034\n\024max_tables_in_sele"
-    "ct\030\036 \001(\r\022\034\n\024max_user_name_length\030\037 \001(\r\0220"
-    "\n\016null_collation\030  \001(\0162\030.exec.user.NullC"
-    "ollation\022&\n\036null_plus_non_null_equals_nu"
-    "ll\030! \001(\010\022\031\n\021numeric_functions\030\" \003(\t\0223\n\020o"
-    "rder_by_support\030# \003(\0162\031.exec.user.OrderB"
-    "ySupport\0227\n\022outer_join_support\030$ \003(\0162\033.e"
-    "xec.user.OuterJoinSupport\022=\n\030quoted_iden"
-    "tifier_casing\030% \001(\0162\033.exec.user.Identifi"
-    "erCasing\022\021\n\tread_only\030& \001(\010\022\023\n\013schema_te"
-    "rm\030\' \001(\t\022\034\n\024search_escape_string\030( \001(\t\022#"
-    "\n\033select_for_update_supported\030) \001(\010\022\032\n\022s"
-    "pecial_characters\030* \001(\t\022\024\n\014sql_keywords\030"
-    "+ \003(\t\022\030\n\020string_functions\030, \003(\t\0224\n\020subqu"
-    "ery_support\030- \003(\0162\032.exec.user.SubQuerySu"
-    "pport\022\030\n\020system_functions\030. \003(\t\022\022\n\ntable"
-    "_term\030/ \001(\t\022\035\n\025transaction_supported\0300 \001"
-    "(\010\022.\n\runion_support\0301 \003(\0162\027.exec.user.Un"
-    "ionSupport\"\353\001\n\010RunQuery\0221\n\014results_mode\030"
-    "\001 \001(\0162\033.exec.user.QueryResultsMode\022$\n\004ty"
-    "pe\030\002 \001(\0162\026.exec.shared.QueryType\022\014\n\004plan"
-    "\030\003 \001(\t\0221\n\tfragments\030\004 \003(\0132\036.exec.bit.con"
-    "trol.PlanFragment\022E\n\031prepared_statement_"
-    "handle\030\005 \001(\0132\".exec.user.PreparedStateme"
-    "ntHandle*\320\003\n\007RpcType\022\r\n\tHANDSHAKE\020\000\022\007\n\003A"
-    "CK\020\001\022\013\n\007GOODBYE\020\002\022\r\n\tRUN_QUERY\020\003\022\020\n\014CANC"
-    "EL_QUERY\020\004\022\023\n\017REQUEST_RESULTS\020\005\022\027\n\023RESUM"
-    "E_PAUSED_QUERY\020\013\022\034\n\030GET_QUERY_PLAN_FRAGM"
-    "ENTS\020\014\022\020\n\014GET_CATALOGS\020\016\022\017\n\013GET_SCHEMAS\020"
-    "\017\022\016\n\nGET_TABLES\020\020\022\017\n\013GET_COLUMNS\020\021\022\035\n\031CR"
-    "EATE_PREPARED_STATEMENT\020\026\022\023\n\017GET_SERVER_"
-    "META\020\010\022\016\n\nQUERY_DATA\020\006\022\020\n\014QUERY_HANDLE\020\007"
-    "\022\030\n\024QUERY_PLAN_FRAGMENTS\020\r\022\014\n\010CATALOGS\020\022"
-    "\022\013\n\007SCHEMAS\020\023\022\n\n\006TABLES\020\024\022\013\n\007COLUMNS\020\025\022\026"
-    "\n\022PREPARED_STATEMENT\020\027\022\017\n\013SERVER_META\020\t\022"
-    "\020\n\014QUERY_RESULT\020\n\022\020\n\014SASL_MESSAGE\020\030*6\n\013S"
-    "aslSupport\022\030\n\024UNKNOWN_SASL_SUPPORT\020\000\022\r\n\t"
-    "SASL_AUTH\020\001*#\n\020QueryResultsMode\022\017\n\013STREA"
-    "M_FULL\020\001*q\n\017HandshakeStatus\022\013\n\007SUCCESS\020\001"
-    "\022\030\n\024RPC_VERSION_MISMATCH\020\002\022\017\n\013AUTH_FAILE"
-    "D\020\003\022\023\n\017UNKNOWN_FAILURE\020\004\022\021\n\rAUTH_REQUIRE"
-    "D\020\005*D\n\rRequestStatus\022\022\n\016UNKNOWN_STATUS\020\000"
-    "\022\006\n\002OK\020\001\022\n\n\006FAILED\020\002\022\013\n\007TIMEOUT\020\003*Y\n\023Col"
-    "umnSearchability\022\031\n\025UNKNOWN_SEARCHABILIT"
-    "Y\020\000\022\010\n\004NONE\020\001\022\010\n\004CHAR\020\002\022\n\n\006NUMBER\020\003\022\007\n\003A"
-    "LL\020\004*K\n\022ColumnUpdatability\022\030\n\024UNKNOWN_UP"
-    "DATABILITY\020\000\022\r\n\tREAD_ONLY\020\001\022\014\n\010WRITABLE\020"
-    "\002*1\n\016CollateSupport\022\016\n\nCS_UNKNOWN\020\000\022\017\n\013C"
-    "S_GROUP_BY\020\001*J\n\027CorrelationNamesSupport\022"
-    "\013\n\007CN_NONE\020\001\022\026\n\022CN_DIFFERENT_NAMES\020\002\022\n\n\006"
-    "CN_ANY\020\003*\271\003\n\027DateTimeLiteralsSupport\022\016\n\n"
-    "DL_UNKNOWN\020\000\022\013\n\007DL_DATE\020\001\022\013\n\007DL_TIME\020\002\022\020"
-    "\n\014DL_TIMESTAMP\020\003\022\024\n\020DL_INTERVAL_YEAR\020\004\022\025"
-    "\n\021DL_INTERVAL_MONTH\020\005\022\023\n\017DL_INTERVAL_DAY"
-    "\020\006\022\024\n\020DL_INTERVAL_HOUR\020\007\022\026\n\022DL_INTERVAL_"
-    "MINUTE\020\010\022\026\n\022DL_INTERVAL_SECOND\020\t\022\035\n\031DL_I"
-    "NTERVAL_YEAR_TO_MONTH\020\n\022\033\n\027DL_INTERVAL_D"
-    "AY_TO_HOUR\020\013\022\035\n\031DL_INTERVAL_DAY_TO_MINUT"
-    "E\020\014\022\035\n\031DL_INTERVAL_DAY_TO_SECOND\020\r\022\036\n\032DL"
-    "_INTERVAL_HOUR_TO_MINUTE\020\016\022\036\n\032DL_INTERVA"
-    "L_HOUR_TO_SECOND\020\017\022 \n\034DL_INTERVAL_MINUTE"
-    "_TO_SECOND\020\020*Y\n\016GroupBySupport\022\013\n\007GB_NON"
-    "E\020\001\022\022\n\016GB_SELECT_ONLY\020\002\022\024\n\020GB_BEYOND_SEL"
-    "ECT\020\003\022\020\n\014GB_UNRELATED\020\004*x\n\020IdentifierCas"
-    "ing\022\016\n\nIC_UNKNOWN\020\000\022\023\n\017IC_STORES_LOWER\020\001"
-    "\022\023\n\017IC_STORES_MIXED\020\002\022\023\n\017IC_STORES_UPPER"
-    "\020\003\022\025\n\021IC_SUPPORTS_MIXED\020\004*X\n\rNullCollati"
-    "on\022\016\n\nNC_UNKNOWN\020\000\022\017\n\013NC_AT_START\020\001\022\r\n\tN"
-    "C_AT_END\020\002\022\013\n\007NC_HIGH\020\003\022\n\n\006NC_LOW\020\004*E\n\016O"
-    "rderBySupport\022\016\n\nOB_UNKNOWN\020\000\022\020\n\014OB_UNRE"
-    "LATED\020\001\022\021\n\rOB_EXPRESSION\020\002*\226\001\n\020OuterJoin"
-    "Support\022\016\n\nOJ_UNKNOWN\020\000\022\013\n\007OJ_LEFT\020\001\022\014\n\010"
-    "OJ_RIGHT\020\002\022\013\n\007OJ_FULL\020\003\022\r\n\tOJ_NESTED\020\004\022\022"
-    "\n\016OJ_NOT_ORDERED\020\005\022\014\n\010OJ_INNER\020\006\022\031\n\025OJ_A"
-    "LL_COMPARISON_OPS\020\007*\204\001\n\017SubQuerySupport\022"
-    "\016\n\nSQ_UNKNOWN\020\000\022\021\n\rSQ_CORRELATED\020\001\022\024\n\020SQ"
-    "_IN_COMPARISON\020\002\022\020\n\014SQ_IN_EXISTS\020\003\022\020\n\014SQ"
-    "_IN_INSERT\020\004\022\024\n\020SQ_IN_QUANTIFIED\020\005*;\n\014Un"
-    "ionSupport\022\r\n\tU_UNKNOWN\020\000\022\013\n\007U_UNION\020\001\022\017"
-    "\n\013U_UNION_ALL\020\002B+\n\033org.apache.drill.exec"
-    ".protoB\nUserProtosH\001", 8820);
+    "exec.user.RpcType\022\021\n\tencrypted\030\t \001(\010\022\026\n\016"
+    "maxWrappedSize\030\n \001(\005\"-\n\nLikeFilter\022\017\n\007pa"
+    "ttern\030\001 \001(\t\022\016\n\006escape\030\002 \001(\t\"D\n\016GetCatalo"
+    "gsReq\0222\n\023catalog_name_filter\030\001 \001(\0132\025.exe"
+    "c.user.LikeFilter\"M\n\017CatalogMetadata\022\024\n\014"
+    "catalog_name\030\001 \001(\t\022\023\n\013description\030\002 \001(\t\022"
+    "\017\n\007connect\030\003 \001(\t\"\223\001\n\017GetCatalogsResp\022(\n\006"
+    "status\030\001 \001(\0162\030.exec.user.RequestStatus\022,"
+    "\n\010catalogs\030\002 \003(\0132\032.exec.user.CatalogMeta"
+    "data\022(\n\005error\030\003 \001(\0132\031.exec.shared.DrillP"
+    "BError\"v\n\rGetSchemasReq\0222\n\023catalog_name_"
+    "filter\030\001 \001(\0132\025.exec.user.LikeFilter\0221\n\022s"
+    "chema_name_filter\030\002 \001(\0132\025.exec.user.Like"
+    "Filter\"i\n\016SchemaMetadata\022\024\n\014catalog_name"
+    "\030\001 \001(\t\022\023\n\013schema_name\030\002 \001(\t\022\r\n\005owner\030\003 \001"
+    "(\t\022\014\n\004type\030\004 \001(\t\022\017\n\007mutable\030\005 \001(\t\"\220\001\n\016Ge"
+    "tSchemasResp\022(\n\006status\030\001 \001(\0162\030.exec.user"
+    ".RequestStatus\022*\n\007schemas\030\002 \003(\0132\031.exec.u"
+    "ser.SchemaMetadata\022(\n\005error\030\003 \001(\0132\031.exec"
+    ".shared.DrillPBError\"\302\001\n\014GetTablesReq\0222\n"
+    "\023catalog_name_filter\030\001 \001(\0132\025.exec.user.L"
+    "ikeFilter\0221\n\022schema_name_filter\030\002 \001(\0132\025."
+    "exec.user.LikeFilter\0220\n\021table_name_filte"
+    "r\030\003 \001(\0132\025.exec.user.LikeFilter\022\031\n\021table_"
+    "type_filter\030\004 \003(\t\"\\\n\rTableMetadata\022\024\n\014ca"
+    "talog_name\030\001 \001(\t\022\023\n\013schema_name\030\002 \001(\t\022\022\n"
+    "\ntable_name\030\003 \001(\t\022\014\n\004type\030\004 \001(\t\"\215\001\n\rGetT"
+    "ablesResp\022(\n\006status\030\001 \001(\0162\030.exec.user.Re"
+    "questStatus\022(\n\006tables\030\002 \003(\0132\030.exec.user."
+    "TableMetadata\022(\n\005error\030\003 \001(\0132\031.exec.shar"
+    "ed.DrillPBError\"\333\001\n\rGetColumnsReq\0222\n\023cat"
+    "alog_name_filter\030\001 \001(\0132\025.exec.user.LikeF"
+    "ilter\0221\n\022schema_name_filter\030\002 \001(\0132\025.exec"
+    ".user.LikeFilter\0220\n\021table_name_filter\030\003 "
+    "\001(\0132\025.exec.user.LikeFilter\0221\n\022column_nam"
+    "e_filter\030\004 \001(\0132\025.exec.user.LikeFilter\"\251\003"
+    "\n\016ColumnMetadata\022\024\n\014catalog_name\030\001 \001(\t\022\023"
+    "\n\013schema_name\030\002 \001(\t\022\022\n\ntable_name\030\003 \001(\t\022"
+    "\023\n\013column_name\030\004 \001(\t\022\030\n\020ordinal_position"
+    "\030\005 \001(\005\022\025\n\rdefault_value\030\006 \001(\t\022\023\n\013is_null"
+    "able\030\007 \001(\010\022\021\n\tdata_type\030\010 \001(\t\022\027\n\017char_ma"
+    "x_length\030\t \001(\005\022\031\n\021char_octet_length\030\n \001("
+    "\005\022\031\n\021numeric_precision\030\013 \001(\005\022\037\n\027numeric_"
+    "precision_radix\030\014 \001(\005\022\025\n\rnumeric_scale\030\r"
+    " \001(\005\022\033\n\023date_time_precision\030\016 \001(\005\022\025\n\rint"
+    "erval_type\030\017 \001(\t\022\032\n\022interval_precision\030\020"
+    " \001(\005\022\023\n\013column_size\030\021 \001(\005\"\220\001\n\016GetColumns"
+    "Resp\022(\n\006status\030\001 \001(\0162\030.exec.user.Request"
+    "Status\022*\n\007columns\030\002 \003(\0132\031.exec.user.Colu"
+    "mnMetadata\022(\n\005error\030\003 \001(\0132\031.exec.shared."
+    "DrillPBError\"/\n\032CreatePreparedStatementR"
+    "eq\022\021\n\tsql_query\030\001 \001(\t\"\326\003\n\024ResultColumnMe"
+    "tadata\022\024\n\014catalog_name\030\001 \001(\t\022\023\n\013schema_n"
+    "ame\030\002 \001(\t\022\022\n\ntable_name\030\003 \001(\t\022\023\n\013column_"
+    "name\030\004 \001(\t\022\r\n\005label\030\005 \001(\t\022\021\n\tdata_type\030\006"
+    " \001(\t\022\023\n\013is_nullable\030\007 \001(\010\022\021\n\tprecision\030\010"
+    " \001(\005\022\r\n\005scale\030\t \001(\005\022\016\n\006signed\030\n \001(\010\022\024\n\014d"
+    "isplay_size\030\013 \001(\005\022\022\n\nis_aliased\030\014 \001(\010\0225\n"
+    "\rsearchability\030\r \001(\0162\036.exec.user.ColumnS"
+    "earchability\0223\n\014updatability\030\016 \001(\0162\035.exe"
+    "c.user.ColumnUpdatability\022\026\n\016auto_increm"
+    "ent\030\017 \001(\010\022\030\n\020case_sensitivity\030\020 \001(\010\022\020\n\010s"
+    "ortable\030\021 \001(\010\022\022\n\nclass_name\030\022 \001(\t\022\023\n\013is_"
+    "currency\030\024 \001(\010\".\n\027PreparedStatementHandl"
+    "e\022\023\n\013server_info\030\001 \001(\014\"\200\001\n\021PreparedState"
+    "ment\0220\n\007columns\030\001 \003(\0132\037.exec.user.Result"
+    "ColumnMetadata\0229\n\rserver_handle\030\002 \001(\0132\"."
+    "exec.user.PreparedStatementHandle\"\253\001\n\033Cr"
+    "eatePreparedStatementResp\022(\n\006status\030\001 \001("
+    "\0162\030.exec.user.RequestStatus\0228\n\022prepared_"
+    "statement\030\002 \001(\0132\034.exec.user.PreparedStat"
+    "ement\022(\n\005error\030\003 \001(\0132\031.exec.shared.Drill"
+    "PBError\"\022\n\020GetServerMetaReq\"P\n\016ConvertSu"
+    "pport\022\037\n\004from\030\001 \002(\0162\021.common.MinorType\022\035"
+    "\n\002to\030\002 \002(\0162\021.common.MinorType\"\223\001\n\021GetSer"
+    "verMetaResp\022(\n\006status\030\001 \001(\0162\030.exec.user."
+    "RequestStatus\022*\n\013server_meta\030\002 \001(\0132\025.exe"
+    "c.user.ServerMeta\022(\n\005error\030\003 \001(\0132\031.exec."
+    "shared.DrillPBError\"\377\r\n\nServerMeta\022\035\n\025al"
+    "l_tables_selectable\030\001 \001(\010\022%\n\035blob_includ"
+    "ed_in_max_row_size\030\002 \001(\010\022\030\n\020catalog_at_s"
+    "tart\030\003 \001(\010\022\031\n\021catalog_separator\030\004 \001(\t\022\024\n"
+    "\014catalog_term\030\005 \001(\t\0222\n\017collate_support\030\006"
+    " \003(\0162\031.exec.user.CollateSupport\022!\n\031colum"
+    "n_aliasing_supported\030\007 \001(\010\0222\n\017convert_su"
+    "pport\030\010 \003(\0132\031.exec.user.ConvertSupport\022E"
+    "\n\031correlation_names_support\030\t \001(\0162\".exec"
+    ".user.CorrelationNamesSupport\022\033\n\023date_ti"
+    "me_functions\030\n \003(\t\022F\n\032date_time_literals"
+    "_support\030\013 \003(\0162\".exec.user.DateTimeLiter"
+    "alsSupport\0223\n\020group_by_support\030\014 \001(\0162\031.e"
+    "xec.user.GroupBySupport\0226\n\021identifier_ca"
+    "sing\030\r \001(\0162\033.exec.user.IdentifierCasing\022"
+    "\037\n\027identifier_quote_string\030\016 \001(\t\022$\n\034like"
+    "_escape_clause_supported\030\017 \001(\010\022!\n\031max_bi"
+    "nary_literal_length\030\020 \001(\r\022\037\n\027max_catalog"
+    "_name_length\030\021 \001(\r\022\037\n\027max_char_literal_l"
+    "ength\030\022 \001(\r\022\036\n\026max_column_name_length\030\023 "
+    "\001(\r\022\037\n\027max_columns_in_group_by\030\024 \001(\r\022\037\n\027"
+    "max_columns_in_order_by\030\025 \001(\r\022\035\n\025max_col"
+    "umns_in_select\030\026 \001(\r\022\036\n\026max_cursor_name_"
+    "length\030\027 \001(\r\022\034\n\024max_logical_lob_size\030\030 \001"
+    "(\r\022\024\n\014max_row_size\030\031 \001(\r\022\036\n\026max_schema_n"
+    "ame_length\030\032 \001(\r\022\034\n\024max_statement_length"
+    "\030\033 \001(\r\022\026\n\016max_statements\030\034 \001(\r\022\035\n\025max_ta"
+    "ble_name_length\030\035 \001(\r\022\034\n\024max_tables_in_s"
+    "elect\030\036 \001(\r\022\034\n\024max_user_name_length\030\037 \001("
+    "\r\0220\n\016null_collation\030  \001(\0162\030.exec.user.Nu"
+    "llCollation\022&\n\036null_plus_non_null_equals"
+    "_null\030! \001(\010\022\031\n\021numeric_functions\030\" \003(\t\0223"
+    "\n\020order_by_support\030# \003(\0162\031.exec.user.Ord"
+    "erBySupport\0227\n\022outer_join_support\030$ \003(\0162"
+    "\033.exec.user.OuterJoinSupport\022=\n\030quoted_i"
+    "dentifier_casing\030% \001(\0162\033.exec.user.Ident"
+    "ifierCasing\022\021\n\tread_only\030& \001(\010\022\023\n\013schema"
+    "_term\030\' \001(\t\022\034\n\024search_escape_string\030( \001("
+    "\t\022#\n\033select_for_update_supported\030) \001(\010\022\032"
+    "\n\022special_characters\030* \001(\t\022\024\n\014sql_keywor"
+    "ds\030+ \003(\t\022\030\n\020string_functions\030, \003(\t\0224\n\020su"
+    "bquery_support\030- \003(\0162\032.exec.user.SubQuer"
+    "ySupport\022\030\n\020system_functions\030. \003(\t\022\022\n\nta"
+    "ble_term\030/ \001(\t\022\035\n\025transaction_supported\030"
+    "0 \001(\010\022.\n\runion_support\0301 \003(\0162\027.exec.user"
+    ".UnionSupport\"\353\001\n\010RunQuery\0221\n\014results_mo"
+    "de\030\001 \001(\0162\033.exec.user.QueryResultsMode\022$\n"
+    "\004type\030\002 \001(\0162\026.exec.shared.QueryType\022\014\n\004p"
+    "lan\030\003 \001(\t\0221\n\tfragments\030\004 \003(\0132\036.exec.bit."
+    "control.PlanFragment\022E\n\031prepared_stateme"
+    "nt_handle\030\005 \001(\0132\".exec.user.PreparedStat"
+    "ementHandle*\320\003\n\007RpcType\022\r\n\tHANDSHAKE\020\000\022\007"
+    "\n\003ACK\020\001\022\013\n\007GOODBYE\020\002\022\r\n\tRUN_QUERY\020\003\022\020\n\014C"
+    "ANCEL_QUERY\020\004\022\023\n\017REQUEST_RESULTS\020\005\022\027\n\023RE"
+    "SUME_PAUSED_QUERY\020\013\022\034\n\030GET_QUERY_PLAN_FR"
+    "AGMENTS\020\014\022\020\n\014GET_CATALOGS\020\016\022\017\n\013GET_SCHEM"
+    "AS\020\017\022\016\n\nGET_TABLES\020\020\022\017\n\013GET_COLUMNS\020\021\022\035\n"
+    "\031CREATE_PREPARED_STATEMENT\020\026\022\023\n\017GET_SERV"
+    "ER_META\020\010\022\016\n\nQUERY_DATA\020\006\022\020\n\014QUERY_HANDL"
+    "E\020\007\022\030\n\024QUERY_PLAN_FRAGMENTS\020\r\022\014\n\010CATALOG"
+    "S\020\022\022\013\n\007SCHEMAS\020\023\022\n\n\006TABLES\020\024\022\013\n\007COLUMNS\020"
+    "\025\022\026\n\022PREPARED_STATEMENT\020\027\022\017\n\013SERVER_META"
+    "\020\t\022\020\n\014QUERY_RESULT\020\n\022\020\n\014SASL_MESSAGE\020\030*H"
+    "\n\013SaslSupport\022\030\n\024UNKNOWN_SASL_SUPPORT\020\000\022"
+    "\r\n\tSASL_AUTH\020\001\022\020\n\014SASL_PRIVACY\020\002*#\n\020Quer"
+    "yResultsMode\022\017\n\013STREAM_FULL\020\001*q\n\017Handsha"
+    "keStatus\022\013\n\007SUCCESS\020\001\022\030\n\024RPC_VERSION_MIS"
+    "MATCH\020\002\022\017\n\013AUTH_FAILED\020\003\022\023\n\017UNKNOWN_FAIL"
+    "URE\020\004\022\021\n\rAUTH_REQUIRED\020\005*D\n\rRequestStatu"
+    "s\022\022\n\016UNKNOWN_STATUS\020\000\022\006\n\002OK\020\001\022\n\n\006FAILED\020"
+    "\002\022\013\n\007TIMEOUT\020\003*Y\n\023ColumnSearchability\022\031\n"
+    "\025UNKNOWN_SEARCHABILITY\020\000\022\010\n\004NONE\020\001\022\010\n\004CH"
+    "AR\020\002\022\n\n\006NUMBER\020\003\022\007\n\003ALL\020\004*K\n\022ColumnUpdat"
+    "ability\022\030\n\024UNKNOWN_UPDATABILITY\020\000\022\r\n\tREA"
+    "D_ONLY\020\001\022\014\n\010WRITABLE\020\002*1\n\016CollateSupport"
+    "\022\016\n\nCS_UNKNOWN\020\000\022\017\n\013CS_GROUP_BY\020\001*J\n\027Cor"
+    "relationNamesSupport\022\013\n\007CN_NONE\020\001\022\026\n\022CN_"
+    "DIFFERENT_NAMES\020\002\022\n\n\006CN_ANY\020\003*\271\003\n\027DateTi"
+    "meLiteralsSupport\022\016\n\nDL_UNKNOWN\020\000\022\013\n\007DL_"
+    "DATE\020\001\022\013\n\007DL_TIME\020\002\022\020\n\014DL_TIMESTAMP\020\003\022\024\n"
+    "\020DL_INTERVAL_YEAR\020\004\022\025\n\021DL_INTERVAL_MONTH"
+    "\020\005\022\023\n\017DL_INTERVAL_DAY\020\006\022\024\n\020DL_INTERVAL_H"
+    "OUR\020\007\022\026\n\022DL_INTERVAL_MINUTE\020\010\022\026\n\022DL_INTE"
+    "RVAL_SECOND\020\t\022\035\n\031DL_INTERVAL_YEAR_TO_MON"
+    "TH\020\n\022\033\n\027DL_INTERVAL_DAY_TO_HOUR\020\013\022\035\n\031DL_"
+    "INTERVAL_DAY_TO_MINUTE\020\014\022\035\n\031DL_INTERVAL_"
+    "DAY_TO_SECOND\020\r\022\036\n\032DL_INTERVAL_HOUR_TO_M"
+    "INUTE\020\016\022\036\n\032DL_INTERVAL_HOUR_TO_SECOND\020\017\022"
+    " \n\034DL_INTERVAL_MINUTE_TO_SECOND\020\020*Y\n\016Gro"
+    "upBySupport\022\013\n\007GB_NONE\020\001\022\022\n\016GB_SELECT_ON"
+    "LY\020\002\022\024\n\020GB_BEYOND_SELECT\020\003\022\020\n\014GB_UNRELAT"
+    "ED\020\004*x\n\020IdentifierCasing\022\016\n\nIC_UNKNOWN\020\000"
+    "\022\023\n\017IC_STORES_LOWER\020\001\022\023\n\017IC_STORES_MIXED"
+    "\020\002\022\023\n\017IC_STORES_UPPER\020\003\022\025\n\021IC_SUPPORTS_M"
+    "IXED\020\004*X\n\rNullCollation\022\016\n\nNC_UNKNOWN\020\000\022"
+    "\017\n\013NC_AT_START\020\001\022\r\n\tNC_AT_END\020\002\022\013\n\007NC_HI"
+    "GH\020\003\022\n\n\006NC_LOW\020\004*E\n\016OrderBySupport\022\016\n\nOB"
+    "_UNKNOWN\020\000\022\020\n\014OB_UNRELATED\020\001\022\021\n\rOB_EXPRE"
+    "SSION\020\002*\226\001\n\020OuterJoinSupport\022\016\n\nOJ_UNKNO"
+    "WN\020\000\022\013\n\007OJ_LEFT\020\001\022\014\n\010OJ_RIGHT\020\002\022\013\n\007OJ_FU"
+    "LL\020\003\022\r\n\tOJ_NESTED\020\004\022\022\n\016OJ_NOT_ORDERED\020\005\022"
+    "\014\n\010OJ_INNER\020\006\022\031\n\025OJ_ALL_COMPARISON_OPS\020\007"
+    "*\204\001\n\017SubQuerySupport\022\016\n\nSQ_UNKNOWN\020\000\022\021\n\r"
+    "SQ_CORRELATED\020\001\022\024\n\020SQ_IN_COMPARISON\020\002\022\020\n"
+    "\014SQ_IN_EXISTS\020\003\022\020\n\014SQ_IN_INSERT\020\004\022\024\n\020SQ_"
+    "IN_QUANTIFIED\020\005*;\n\014UnionSupport\022\r\n\tU_UNK"
+    "NOWN\020\000\022\013\n\007U_UNION\020\001\022\017\n\013U_UNION_ALL\020\002B+\n\033"
+    "org.apache.drill.exec.protoB\nUserProtosH"
+    "\001", 8881);
   ::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
     "User.proto", &protobuf_RegisterTypes);
   Property::default_instance_ = new Property();
@@ -1262,6 +1266,7 @@ bool SaslSupport_IsValid(int value) {
   switch(value) {
     case 0:
     case 1:
+    case 2:
       return true;
     default:
       return false;
@@ -4044,6 +4049,8 @@ const int BitToUserHandshake::kErrorMessageFieldNumber;
 const int BitToUserHandshake::kServerInfosFieldNumber;
 const int BitToUserHandshake::kAuthenticationMechanismsFieldNumber;
 const int BitToUserHandshake::kSupportedMethodsFieldNumber;
+const int BitToUserHandshake::kEncryptedFieldNumber;
+const int BitToUserHandshake::kMaxWrappedSizeFieldNumber;
 #endif  // !_MSC_VER
 
 BitToUserHandshake::BitToUserHandshake()
@@ -4068,6 +4075,8 @@ void BitToUserHandshake::SharedCtor() {
   errorid_ = const_cast< ::std::string*>(&::google::protobuf::internal::kEmptyString);
   errormessage_ = const_cast< ::std::string*>(&::google::protobuf::internal::kEmptyString);
   server_infos_ = NULL;
+  encrypted_ = false;
+  maxwrappedsize_ = 0;
   ::memset(_has_bits_, 0, sizeof(_has_bits_));
 }
 
@@ -4125,6 +4134,10 @@ void BitToUserHandshake::Clear() {
     if (has_server_infos()) {
       if (server_infos_ != NULL) server_infos_->::exec::user::RpcEndpointInfos::Clear();
     }
+    encrypted_ = false;
+  }
+  if (_has_bits_[8 / 32] & (0xffu << (8 % 32))) {
+    maxwrappedsize_ = 0;
   }
   authenticationmechanisms_.Clear();
   supported_methods_.Clear();
@@ -4266,6 +4279,38 @@ bool BitToUserHandshake::MergePartialFromCodedStream(
           goto handle_uninterpreted;
         }
         if (input->ExpectTag(64)) goto parse_supported_methods;
+        if (input->ExpectTag(72)) goto parse_encrypted;
+        break;
+      }
+
+      // optional bool encrypted = 9;
+      case 9: {
+        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
+         parse_encrypted:
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+                   bool, ::google::protobuf::internal::WireFormatLite::TYPE_BOOL>(
+                 input, &encrypted_)));
+          set_has_encrypted();
+        } else {
+          goto handle_uninterpreted;
+        }
+        if (input->ExpectTag(80)) goto parse_maxWrappedSize;
+        break;
+      }
+
+      // optional int32 maxWrappedSize = 10;
+      case 10: {
+        if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
+            ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
+         parse_maxWrappedSize:
+          DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+                   ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
+                 input, &maxwrappedsize_)));
+          set_has_maxwrappedsize();
+        } else {
+          goto handle_uninterpreted;
+        }
         if (input->ExpectAtEnd()) return true;
         break;
       }
@@ -4338,6 +4383,16 @@ void BitToUserHandshake::SerializeWithCachedSizes(
       8, this->supported_methods(i), output);
   }
 
+  // optional bool encrypted = 9;
+  if (has_encrypted()) {
+    ::google::protobuf::internal::WireFormatLite::WriteBool(9, this->encrypted(), output);
+  }
+
+  // optional int32 maxWrappedSize = 10;
+  if (has_maxwrappedsize()) {
+    ::google::protobuf::internal::WireFormatLite::WriteInt32(10, this->maxwrappedsize(), output);
+  }
+
   if (!unknown_fields().empty()) {
     ::google::protobuf::internal::WireFormat::SerializeUnknownFields(
         unknown_fields(), output);
@@ -4399,6 +4454,16 @@ void BitToUserHandshake::SerializeWithCachedSizes(
       8, this->supported_methods(i), target);
   }
 
+  // optional bool encrypted = 9;
+  if (has_encrypted()) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteBoolToArray(9, this->encrypted(), target);
+  }
+
+  // optional int32 maxWrappedSize = 10;
+  if (has_maxwrappedsize()) {
+    target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(10, this->maxwrappedsize(), target);
+  }
+
   if (!unknown_fields().empty()) {
     target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
         unknown_fields(), target);
@@ -4444,6 +4509,20 @@ int BitToUserHandshake::ByteSize() const {
           this->server_infos());
     }
 
+    // optional bool encrypted = 9;
+    if (has_encrypted()) {
+      total_size += 1 + 1;
+    }
+
+  }
+  if (_has_bits_[8 / 32] & (0xffu << (8 % 32))) {
+    // optional int32 maxWrappedSize = 10;
+    if (has_maxwrappedsize()) {
+      total_size += 1 +
+        ::google::protobuf::internal::WireFormatLite::Int32Size(
+          this->maxwrappedsize());
+    }
+
   }
   // repeated string authenticationMechanisms = 7;
   total_size += 1 * this->authenticationmechanisms_size();
@@ -4505,6 +4584,14 @@ void BitToUserHandshake::MergeFrom(const BitToUserHandshake& from) {
     if (from.has_server_infos()) {
       mutable_server_infos()->::exec::user::RpcEndpointInfos::MergeFrom(from.server_infos());
     }
+    if (from.has_encrypted()) {
+      set_encrypted(from.encrypted());
+    }
+  }
+  if (from._has_bits_[8 / 32] & (0xffu << (8 % 32))) {
+    if (from.has_maxwrappedsize()) {
+      set_maxwrappedsize(from.maxwrappedsize());
+    }
   }
   mutable_unknown_fields()->MergeFrom(from.unknown_fields());
 }
@@ -4535,6 +4622,8 @@ void BitToUserHandshake::Swap(BitToUserHandshake* other) {
     std::swap(server_infos_, other->server_infos_);
     authenticationmechanisms_.Swap(&other->authenticationmechanisms_);
     supported_methods_.Swap(&other->supported_methods_);
+    std::swap(encrypted_, other->encrypted_);
+    std::swap(maxwrappedsize_, other->maxwrappedsize_);
     std::swap(_has_bits_[0], other->_has_bits_[0]);
     _unknown_fields_.Swap(&other->_unknown_fields_);
     std::swap(_cached_size_, other->_cached_size_);
diff --git a/contrib/native/client/src/protobuf/User.pb.h b/contrib/native/client/src/protobuf/User.pb.h
index a8f8db4d3..b50e8c386 100644
--- a/contrib/native/client/src/protobuf/User.pb.h
+++ b/contrib/native/client/src/protobuf/User.pb.h
@@ -117,11 +117,12 @@ inline bool RpcType_Parse(
 }
 enum SaslSupport {
   UNKNOWN_SASL_SUPPORT = 0,
-  SASL_AUTH = 1
+  SASL_AUTH = 1,
+  SASL_PRIVACY = 2
 };
 bool SaslSupport_IsValid(int value);
 const SaslSupport SaslSupport_MIN = UNKNOWN_SASL_SUPPORT;
-const SaslSupport SaslSupport_MAX = SASL_AUTH;
+const SaslSupport SaslSupport_MAX = SASL_PRIVACY;
 const int SaslSupport_ARRAYSIZE = SaslSupport_MAX + 1;
 
 const ::google::protobuf::EnumDescriptor* SaslSupport_descriptor();
@@ -1439,6 +1440,20 @@ class BitToUserHandshake : public ::google::protobuf::Message {
   inline const ::google::protobuf::RepeatedField<int>& supported_methods() const;
   inline ::google::protobuf::RepeatedField<int>* mutable_supported_methods();
 
+  // optional bool encrypted = 9;
+  inline bool has_encrypted() const;
+  inline void clear_encrypted();
+  static const int kEncryptedFieldNumber = 9;
+  inline bool encrypted() const;
+  inline void set_encrypted(bool value);
+
+  // optional int32 maxWrappedSize = 10;
+  inline bool has_maxwrappedsize() const;
+  inline void clear_maxwrappedsize();
+  static const int kMaxWrappedSizeFieldNumber = 10;
+  inline ::google::protobuf::int32 maxwrappedsize() const;
+  inline void set_maxwrappedsize(::google::protobuf::int32 value);
+
   // @@protoc_insertion_point(class_scope:exec.user.BitToUserHandshake)
  private:
   inline void set_has_rpc_version();
@@ -1451,6 +1466,10 @@ class BitToUserHandshake : public ::google::protobuf::Message {
   inline void clear_has_errormessage();
   inline void set_has_server_infos();
   inline void clear_has_server_infos();
+  inline void set_has_encrypted();
+  inline void clear_has_encrypted();
+  inline void set_has_maxwrappedsize();
+  inline void clear_has_maxwrappedsize();
 
   ::google::protobuf::UnknownFieldSet _unknown_fields_;
 
@@ -1461,9 +1480,11 @@ class BitToUserHandshake : public ::google::protobuf::Message {
   ::exec::user::RpcEndpointInfos* server_infos_;
   ::google::protobuf::RepeatedPtrField< ::std::string> authenticationmechanisms_;
   ::google::protobuf::RepeatedField<int> supported_methods_;
+  bool encrypted_;
+  ::google::protobuf::int32 maxwrappedsize_;
 
   mutable int _cached_size_;
-  ::google::protobuf::uint32 _has_bits_[(7 + 31) / 32];
+  ::google::protobuf::uint32 _has_bits_[(9 + 31) / 32];
 
   friend void  protobuf_AddDesc_User_2eproto();
   friend void protobuf_AssignDesc_User_2eproto();
@@ -6226,6 +6247,50 @@ BitToUserHandshake::mutable_supported_methods() {
   return &supported_methods_;
 }
 
+// optional bool encrypted = 9;
+inline bool BitToUserHandshake::has_encrypted() const {
+  return (_has_bits_[0] & 0x00000080u) != 0;
+}
+inline void BitToUserHandshake::set_has_encrypted() {
+  _has_bits_[0] |= 0x00000080u;
+}
+inline void BitToUserHandshake::clear_has_encrypted() {
+  _has_bits_[0] &= ~0x00000080u;
+}
+inline void BitToUserHandshake::clear_encrypted() {
+  encrypted_ = false;
+  clear_has_encrypted();
+}
+inline bool BitToUserHandshake::encrypted() const {
+  return encrypted_;
+}
+inline void BitToUserHandshake::set_encrypted(bool value) {
+  set_has_encrypted();
+  encrypted_ = value;
+}
+
+// optional int32 maxWrappedSize = 10;
+inline bool BitToUserHandshake::has_maxwrappedsize() const {
+  return (_has_bits_[0] & 0x00000100u) != 0;
+}
+inline void BitToUserHandshake::set_has_maxwrappedsize() {
+  _has_bits_[0] |= 0x00000100u;
+}
+inline void BitToUserHandshake::clear_has_maxwrappedsize() {
+  _has_bits_[0] &= ~0x00000100u;
+}
+inline void BitToUserHandshake::clear_maxwrappedsize() {
+  maxwrappedsize_ = 0;
+  clear_has_maxwrappedsize();
+}
+inline ::google::protobuf::int32 BitToUserHandshake::maxwrappedsize() const {
+  return maxwrappedsize_;
+}
+inline void BitToUserHandshake::set_maxwrappedsize(::google::protobuf::int32 value) {
+  set_has_maxwrappedsize();
+  maxwrappedsize_ = value;
+}
+
 // -------------------------------------------------------------------
 
 // LikeFilter