Merge branch 'cassandra-3.0' into trunk

CHANGES.txt

@@ -5,6 +5,7 @@
 
 
 3.0
+ * Support encrypted and plain traffic on the same port (CASSANDRA-10559)
  * Fix handling of range tombstones when reading old format sstables (CASSANDRA-10360)
  * Aggregate with Initial Condition fails with C* 3.0 (CASSANDRA-10367)
 Merged from 2.2:

conf/cassandra.yaml

@@ -884,6 +884,8 @@ server_encryption_options:
 # enable or disable client/server encryption.
 client_encryption_options:
     enabled: false
+    # If enabled and optional is set to true encrypted and unencrypted connections are handled.
+    optional: false
     keystore: conf/.keystore
     keystore_password: cassandra
     # require_client_auth: false

src/java/org/apache/cassandra/config/EncryptionOptions.java

@@ -36,6 +36,7 @@ public abstract class EncryptionOptions
     public static class ClientEncryptionOptions extends EncryptionOptions
     {
         public boolean enabled = false;
+        public boolean optional = false;
     }
 
     public static class ServerEncryptionOptions extends EncryptionOptions

src/java/org/apache/cassandra/transport/Server.java

@@ -33,9 +33,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
 import io.netty.channel.*;
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.epoll.EpollServerSocketChannel;
+import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.DefaultChannelGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -139,8 +141,16 @@ public class Server implements CassandraDaemon.Server
         final EncryptionOptions.ClientEncryptionOptions clientEnc = DatabaseDescriptor.getClientEncryptionOptions();
         if (this.useSSL)
         {
-            logger.info("Enabling encrypted CQL connections between client and server");
-            bootstrap.childHandler(new SecureInitializer(this, clientEnc));
+            if (clientEnc.optional)
+            {
+                logger.info("Enabling optionally encrypted CQL connections between client and server");
+                bootstrap.childHandler(new OptionalSecureInitializer(this, clientEnc));
+            }
+            else
+            {
+                logger.info("Enabling encrypted CQL connections between client and server");
+                bootstrap.childHandler(new SecureInitializer(this, clientEnc));
+            }
         }
         else
         {
@@ -326,12 +336,12 @@ public class Server implements CassandraDaemon.Server
         }
     }
 
-    private static class SecureInitializer extends Initializer
+    protected abstract static class AbstractSecureIntializer extends Initializer
     {
         private final SSLContext sslContext;
         private final EncryptionOptions encryptionOptions;
 
-        public SecureInitializer(Server server, EncryptionOptions encryptionOptions)
+        protected AbstractSecureIntializer(Server server, EncryptionOptions encryptionOptions)
         {
             super(server);
             this.encryptionOptions = encryptionOptions;
@@ -345,14 +355,65 @@ public class Server implements CassandraDaemon.Server
             }
         }
 
-        protected void initChannel(Channel channel) throws Exception
-        {
+        protected final SslHandler createSslHandler() {
             SSLEngine sslEngine = sslContext.createSSLEngine();
             sslEngine.setUseClientMode(false);
             sslEngine.setEnabledCipherSuites(encryptionOptions.cipher_suites);
             sslEngine.setNeedClientAuth(encryptionOptions.require_client_auth);
             sslEngine.setEnabledProtocols(SSLFactory.ACCEPTED_PROTOCOLS);
-            SslHandler sslHandler = new SslHandler(sslEngine);
+            return new SslHandler(sslEngine);
+        }
+    }
+
+    private static class OptionalSecureInitializer extends AbstractSecureIntializer
+    {
+        public OptionalSecureInitializer(Server server, EncryptionOptions encryptionOptions)
+        {
+            super(server, encryptionOptions);
+        }
+
+        protected void initChannel(final Channel channel) throws Exception
+        {
+            super.initChannel(channel);
+            channel.pipeline().addFirst("sslDetectionHandler", new ByteToMessageDecoder()
+            {
+                @Override
+                protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception
+                {
+                    if (byteBuf.readableBytes() < 5)
+                    {
+                        // To detect if SSL must be used we need to have at least 5 bytes, so return here and try again
+                        // once more bytes a ready.
+                        return;
+                    }
+                    if (SslHandler.isEncrypted(byteBuf))
+                    {
+                        // Connection uses SSL/TLS, replace the detection handler with a SslHandler and so use
+                        // encryption.
+                        SslHandler sslHandler = createSslHandler();
+                        channelHandlerContext.pipeline().replace(this, "ssl", sslHandler);
+                    }
+                    else
+                    {
+                        // Connection use no TLS/SSL encryption, just remove the detection handler and continue without
+                        // SslHandler in the pipeline.
+                        channelHandlerContext.pipeline().remove(this);
+                    }
+                }
+            });
+        }
+    }
+
+    private static class SecureInitializer extends AbstractSecureIntializer
+    {
+        public SecureInitializer(Server server, EncryptionOptions encryptionOptions)
+        {
+            super(server, encryptionOptions);
+        }
+
+        protected void initChannel(Channel channel) throws Exception
+        {
+            SslHandler sslHandler = createSslHandler();
             super.initChannel(channel);
             channel.pipeline().addFirst("ssl", sslHandler);
         }

test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java

@@ -123,6 +123,24 @@ public class NativeTransportServiceTest
     {
         // default ssl settings: client encryption enabled and default native transport port used for ssl only
         DatabaseDescriptor.getClientEncryptionOptions().enabled = true;
+        DatabaseDescriptor.getClientEncryptionOptions().optional = false;
+
+        withService((NativeTransportService service) ->
+                    {
+                        service.initialize();
+                        assertEquals(1, service.getServers().size());
+                        Server server = service.getServers().iterator().next();
+                        assertTrue(server.useSSL);
+                        assertEquals(server.socket.getPort(), DatabaseDescriptor.getNativeTransportPort());
+                    }, false, 1);
+    }
+
+    @Test
+    public void testSSLOptional()
+    {
+        // default ssl settings: client encryption enabled and default native transport port used for optional ssl
+        DatabaseDescriptor.getClientEncryptionOptions().enabled = true;
+        DatabaseDescriptor.getClientEncryptionOptions().optional = true;
 
         withService((NativeTransportService service) ->
                     {