Don't allow non-secure client connections

(1) Introduced a new config option onlySecureClientsAllowed in
    the Bookie Configuration. Default value is considered false.
(2) If onlySecureClientsAllowed is set to True, the Bookies only
    allow the Clients to communicate over TLS. Any requests
    from the Clients without TLS enabled will be rejected
    and Client gets the Security Exception.

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java

@@ -348,6 +348,9 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
     protected static final String MAX_BATCH_READ_SIZE = "maxBatchReadSize";
     protected static final int DEFAULT_MAX_BATCH_READ_SIZE = 5 * 1024 * 1024; // 5MB
 
+    // Is true if only Secure Client Connections are allowed.
+    protected static final String ONLY_SECURE_CLIENTS_ALLOWED = "onlySecureClientsAllowed";
+
     /**
      * Construct a default configuration object.
      */
@@ -4158,4 +4161,19 @@ private String getFilePath(String fileName) {
         }
         return "";
     }
+
+    /*
+     * True if only secured client connections are allowed.
+     */
+    public boolean isOnlySecureClientConnectionAllowed() {
+        return this.getBoolean(ONLY_SECURE_CLIENTS_ALLOWED, false);
+    }
+
+    /**
+     * Enable/Disable the feature to allow only Secure Client Connections.
+     */
+    public ServerConfiguration setOnlySecureClientConnectionAllowed(boolean onlySecureClientConnectionAllowed) {
+        this.setProperty(ONLY_SECURE_CLIENTS_ALLOWED, Boolean.toString(onlySecureClientConnectionAllowed));
+        return this;
+    }
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/AuthHandler.java

@@ -29,6 +29,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelPromise;
+import io.netty.handler.ssl.SslHandler;
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.util.Queue;
@@ -39,6 +40,7 @@
 import org.apache.bookkeeper.auth.BookieAuthProvider;
 import org.apache.bookkeeper.auth.ClientAuthProvider;
 import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.proto.BookkeeperProtocol.AuthMessage;
 import org.apache.bookkeeper.util.ByteBufList;
 import org.apache.bookkeeper.util.NettyChannelUtil;
@@ -52,9 +54,12 @@ static class ServerSideHandler extends ChannelInboundHandlerAdapter {
         volatile boolean authenticated = false;
         final BookieAuthProvider.Factory authProviderFactory;
         final BookieConnectionPeer connectionPeer;
+        final ServerConfiguration serverCfg;
         BookieAuthProvider authProvider;
 
-        ServerSideHandler(BookieConnectionPeer connectionPeer, BookieAuthProvider.Factory authProviderFactory) {
+        ServerSideHandler(ServerConfiguration serverCfg, BookieConnectionPeer connectionPeer,
+                          BookieAuthProvider.Factory authProviderFactory) {
+            this.serverCfg = serverCfg;
             this.authProviderFactory = authProviderFactory;
             this.connectionPeer = connectionPeer;
             authProvider = null;
@@ -86,6 +91,24 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             }
 
             if (authenticated) {
+                if (msg instanceof BookkeeperProtocol.Request) {
+                    Channel c = ctx.channel();
+                    BookkeeperProtocol.Request req = (BookkeeperProtocol.Request) msg;
+                    // If onlySecureClientsAllowed is set to True, Allow operations from only the Secure Clients.
+                    // The Secure Clients have the tls handler installed in their Channel Pipeline.
+                    if (serverCfg.isOnlySecureClientConnectionAllowed()
+                            && req.getHeader().getOperation() != BookkeeperProtocol.OperationType.START_TLS
+                            && c.pipeline().get(SslHandler.class) == null) {
+                            LOG.error("Request from Non-secure connection is not allowed",
+                                    req.getHeader().getOperation());
+                            // We return Authentication Failure(EAUTH) error to Client
+                            BookkeeperProtocol.Response.Builder builder = BookkeeperProtocol.Response.newBuilder()
+                                    .setHeader(req.getHeader())
+                                    .setStatus(BookkeeperProtocol.StatusCode.EUA);
+                            ctx.channel().writeAndFlush(builder.build());
+                            return;
+                    }
+                }
                 super.channelRead(ctx, msg);
             } else if (msg instanceof BookieProtocol.AuthRequest) { // pre-PB-client
                 BookieProtocol.AuthRequest req = (BookieProtocol.AuthRequest) msg;

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java

@@ -353,7 +353,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
 
                     pipeline.addLast("bookieProtoDecoder", new BookieProtoEncoding.RequestDecoder(registry));
                     pipeline.addLast("bookieProtoEncoder", new BookieProtoEncoding.ResponseEncoder(registry));
-                    pipeline.addLast("bookieAuthHandler", new AuthHandler.ServerSideHandler(
+                    pipeline.addLast("bookieAuthHandler", new AuthHandler.ServerSideHandler(conf,
                                 contextHandler.getConnectionPeer(), authProviderFactory));
 
                     ChannelInboundHandler requestHandler = isRunning.get()
@@ -419,7 +419,7 @@ protected void initChannel(LocalChannel ch) throws Exception {
 
                     pipeline.addLast("bookieProtoDecoder", new BookieProtoEncoding.RequestDecoder(registry));
                     pipeline.addLast("bookieProtoEncoder", new BookieProtoEncoding.ResponseEncoder(registry));
-                    pipeline.addLast("bookieAuthHandler", new AuthHandler.ServerSideHandler(
+                    pipeline.addLast("bookieAuthHandler", new AuthHandler.ServerSideHandler(conf,
                                 contextHandler.getConnectionPeer(), authProviderFactory));
 
                     ChannelInboundHandler requestHandler = isRunning.get()

bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieRequestProcessor.java

@@ -370,6 +370,12 @@ public void processRequest(Object msg, BookieRequestHandler requestHandler) {
                 MDC.clear();
             }
         } else {
+            if (serverCfg.isOnlySecureClientConnectionAllowed()) {
+                LOG.error("Client request of an older protocol version " + BookieProtocol.CURRENT_PROTOCOL_VERSION
+                        +  " denied because server config option \"onlySecureClientsAllowed=true\" and this version "
+                        + "of the protocol does not support TLS.");
+                return;
+            }
             BookieProtocol.Request r = (BookieProtocol.Request) msg;
             // process packet
             switch (r.getOpCode()) {

bookkeeper-server/src/test/java/org/apache/bookkeeper/tls/TestTLS.java

@@ -110,7 +110,7 @@ public static Collection<Object[]> data() {
     public TestTLS(String keyStoreFormat,
                    String trustStoreFormat,
                    boolean useV2Protocol) {
-        super(3);
+        super(1);
         this.clientKeyStoreFormat = KeyStoreType.valueOf(keyStoreFormat);
         this.clientTrustStoreFormat = KeyStoreType.valueOf(trustStoreFormat);
         this.serverKeyStoreFormat = KeyStoreType.valueOf(keyStoreFormat);
@@ -504,9 +504,11 @@ public void testConnectToTLSClusterTLSClientWithAuthentication() throws Exceptio
 
     /**
      * Verify that a client without tls enabled can connect to a cluster with TLS.
+     * In the Default Bookie config, ONLY_SECURE_CLIENTS_ALLOWED is set to false.
+     * Therefore, the Bookie allows non-secure client connection.
      */
     @Test
-    public void testConnectToTLSClusterNonTLSClient() throws Exception {
+    public void testConnectToTLSClusterNonTLSClient1() throws Exception {
         ClientConfiguration conf = new ClientConfiguration(baseClientConf);
         conf.setTLSProviderFactoryClass(null);
         try {
@@ -516,6 +518,86 @@ public void testConnectToTLSClusterNonTLSClient() throws Exception {
         }
     }
 
+    /**
+     * Verify that a client without tls enabled can NOT connect to a cluster with TLS,
+     * if ONLY_SECURE_CLIENTS_ALLOWED is set in the Bookie config.
+     */
+    @Test
+    public void testConnectToTLSClusterNonTLSClient2() throws Exception {
+        // Set client without TLS
+        ClientConfiguration conf = new ClientConfiguration(baseClientConf);
+        conf.setTLSProviderFactoryClass(null);
+        try {
+            // Enable the feature to only allow secure clients.
+            restartBookies(c -> {
+                c.setOnlySecureClientConnectionAllowed(true);
+                return c;
+            });
+            testClient(conf, numBookies);
+            fail("non tls client should not be able to connect to tls enabled bookies");
+        } catch (BKException.BKNotEnoughBookiesException nnbe) {
+            // Correct response.
+        } catch (BKException.BKUnauthorizedAccessException ue) {
+            // Correct response.
+        }
+    }
+
+    /**
+     *  Verify that the Read from the non-Secure Client throws BKSecurityException
+     *  if ONLY_SECURE_CLIENTS_ALLOWED is set in the Bookie config.
+     */
+    @Test
+    public void testReadwithNonTLSBookie() throws Exception {
+
+        /* TestTLS tests for V3 and V2 protocols together
+         * We should be able to create a ledger when its v3 and setOnlySecureClientConnectionAllowed(true)
+         * We shouldn't be able to do any create / read when its v2 and setOnlySecureClientConnectionAllowed(true)
+         * since v2 does not support TLS.
+         */
+        if (useV2Protocol) {
+            return;
+        }
+
+        // Enable the feature to only allow secure clients.
+        ClientConfiguration conf = new ClientConfiguration(baseClientConf);
+        long lid = 0;
+        try {
+            restartBookies(c -> {
+                c.setOnlySecureClientConnectionAllowed(true);
+                return c;
+            });
+            // Create the Ledger
+            BookKeeper client = new BookKeeper(conf);
+            byte[] passwd = "testPassword".getBytes();
+            int numEntries = 10;
+            byte[] testEntry = "testEntry".getBytes();
+            try (LedgerHandle lh = client.createLedger(numBookies, numBookies, DigestType.CRC32, passwd);) {
+                for (int i = 0; i <= numEntries; i++) {
+                    lh.addEntry(testEntry);
+                }
+                lid = lh.getId();
+            }
+        } catch (BKException.BKNotEnoughBookiesException nnbe) {
+            fail("tls client could not create the ledger");
+        }
+        // nonTLS client should not be able to read the ledger
+        conf.setTLSProviderFactoryClass(null);
+        try {
+            BookKeeper client = new BookKeeper(conf);
+            byte[] passwd = "testPassword".getBytes();
+            int numEntries = 10;
+            byte[] testEntry = "testEntry".getBytes();
+            try (LedgerHandle lh = client.openLedger(lid, DigestType.CRC32, passwd);) {
+                Enumeration<LedgerEntry> entries = lh.readEntries(0, numEntries);
+                fail("The Read should've failed with BKSecurityException");
+            }
+        } catch (BKException.BKSecurityException se) {
+            // Correct Response.
+        }  catch (BKException.BKUnauthorizedAccessException ue) {
+            // Correct response.
+        }
+    }
+
     /**
      * Verify that a client will fail to connect to a server if it has asked for TLS, but it is not available.
      */
@@ -540,7 +622,7 @@ public void testClientWantsTLSNoServersHaveIt() throws Exception {
      * bookies with TLS enabled in the cluster, although few bookies do not have TLS enabled.
      */
     @Test
-    public void testTLSClientButOnlyFewTLSServers() throws Exception {
+    public void testTLSClientButOnlyFewTLSServers1() throws Exception {
         // disable TLS on initial set of bookies
         restartBookies(c -> {
                 c.setTLSProviderFactoryClass(null);

conf/bk_server.conf

@@ -300,6 +300,10 @@ httpServerTrustStorePassword=
 # Tls certificate files refresh duration in seconds.
 # tlsCertFilesRefreshDurationSeconds=0
 
+# Set true if only Secure Client Connection is allowed
+# onlySecureClientsAllowed=
+
+
 ############################################## Bookie Storage ##############################################