[SPARK-13331] AES support for over-the-wire encryption

Encrypt the config message since it was removed by mistake.

Author: Junjie Chen

common/network-common/src/main/java/org/apache/spark/network/sasl/SaslClientBootstrap.java

@@ -95,13 +95,20 @@ public void doBootstrap(TransportClient client, Channel channel) {
           // Generate a request config message to send to server.
           AesConfigMessage configMessage = AesCipher.createConfigMessage(conf);
           ByteBuffer buf = configMessage.encodeMessage();
-          client.sendRpcSync(buf, conf.saslRTTimeoutMs());
+
+          // Encrypted the config message.
+          ByteBuffer encrypted = ByteBuffer.wrap(
+            saslClient.wrap(buf.array(), 0, buf.array().length));
+
+          client.sendRpcSync(encrypted, conf.saslRTTimeoutMs());
           AesCipher cipher = new AesCipher(configMessage);
           logger.info("Enabling AES cipher for client channel {}", client);
           cipher.addToChannel(channel);
         } else {
           SaslEncryption.addToChannel(channel, saslClient, conf.maxSaslEncryptedBlockSize());
         }
+
+        saslClient.dispose();
         saslClient = null;
         logger.debug("Channel {} configured for encryption.", client);
       }

common/network-common/src/main/java/org/apache/spark/network/sasl/SaslRpcHandler.java

@@ -139,7 +139,16 @@ public void receive(TransportClient client, ByteBuffer message, RpcResponseCallb
 
       // Create AES cipher when it is authenticated
       try {
-        AesConfigMessage configMessage = AesConfigMessage.decodeMessage(message);
+        byte[] encrypted;
+        if (message.hasArray()) {
+          encrypted = message.array();
+        } else {
+          encrypted = new byte[message.remaining()];
+          message.get(encrypted);
+        }
+        ByteBuffer decrypted = ByteBuffer.wrap(saslServer.unwrap(encrypted, 0 , encrypted.length));
+
+        AesConfigMessage configMessage = AesConfigMessage.decodeMessage(decrypted);
         AesCipher cipher = new AesCipher(configMessage);
 
         // Send response back to client to confirm that server accept config.