Adding support for TLS client authentication for internode communication

patch by Jyothsna Konisa; reviewed by Bernardo Botella, Francisco Guerrero, Jon Meredith, Maulin Vasavada, Yifan Cai for CASSANDRA-17513

CHANGES.txt

@@ -1,4 +1,5 @@
 4.2
+ * Adding support for TLS client authentication for internode communication (CASSANDRA-17513)
  * Add new CQL function maxWritetime (CASSANDRA-17425)
  * Add guardrail for ALTER TABLE ADD / DROP / REMOVE column operations (CASSANDRA-17495)
  * Rename DisableFlag class to EnableFlag on guardrails (CASSANDRA-17544)

conf/cassandra.yaml

@@ -1316,6 +1316,12 @@ server_encryption_options:
   # Set to a valid keystore if internode_encryption is dc, rack or all
   keystore: conf/.keystore
   keystore_password: cassandra
+  # During internode mTLS authentication, inbound connections (acting as servers) use keystore, keystore_password
+  # containing server certificate to create SSLContext and
+  # outbound connections (acting as clients) use outbound_keystore & outbound_keystore_password with client certificates
+  # to create SSLContext. By default, outbound_keystore is the same as keystore indicating mTLS is not enabled.
+#  outbound_keystore: conf/.keystore
+#  outbound_keystore_password: cassandra
   # Verify peer server certificates
   require_client_auth: false
   # Set to a valid trustore if require_client_auth is true

src/java/org/apache/cassandra/net/OutboundConnectionSettings.java

@@ -82,7 +82,7 @@ public static Framing forId(int id)
     public final IInternodeAuthenticator authenticator;
     public final InetAddressAndPort to;
     public final InetAddressAndPort connectTo; // may be represented by a different IP address on this node's local network
-    public final EncryptionOptions encryption;
+    public final ServerEncryptionOptions encryption;
     public final Framing framing;
     public final Integer socketSendBufferSizeInBytes;
     public final Integer applicationSendQueueCapacityInBytes;
@@ -112,7 +112,7 @@ public OutboundConnectionSettings(InetAddressAndPort to, InetAddressAndPort pref
     private OutboundConnectionSettings(IInternodeAuthenticator authenticator,
                                        InetAddressAndPort to,
                                        InetAddressAndPort connectTo,
-                                       EncryptionOptions encryption,
+                                       ServerEncryptionOptions encryption,
                                        Framing framing,
                                        Integer socketSendBufferSizeInBytes,
                                        Integer applicationSendQueueCapacityInBytes,
@@ -365,7 +365,7 @@ public OutboundDebugCallbacks debug()
         return debug != null ? debug : OutboundDebugCallbacks.NONE;
     }
 
-    public EncryptionOptions encryption()
+    public ServerEncryptionOptions encryption()
     {
         return encryption != null ? encryption : defaultEncryptionOptions(to);
     }
@@ -499,7 +499,7 @@ private static boolean isInLocalDC(IEndpointSnitch snitch, InetAddressAndPort lo
     }
 
     @VisibleForTesting
-    static EncryptionOptions defaultEncryptionOptions(InetAddressAndPort endpoint)
+    static ServerEncryptionOptions defaultEncryptionOptions(InetAddressAndPort endpoint)
     {
         ServerEncryptionOptions options = DatabaseDescriptor.getInternodeMessagingEncyptionOptions();
         return options.shouldEncrypt(endpoint) ? options : null;

src/java/org/apache/cassandra/security/AbstractSslContextFactory.java

@@ -178,15 +178,16 @@ public SslContext createNettySslContext(boolean verifyPeerCertificate, SocketTyp
             key file in PEM format (see {@link SslContextBuilder#forServer(File, File, String)}). However, we are
             not supporting that now to keep the config/yaml API simple.
          */
-        KeyManagerFactory kmf = buildKeyManagerFactory();
         SslContextBuilder builder;
         if (socketType == SocketType.SERVER)
         {
+            KeyManagerFactory kmf = buildKeyManagerFactory();
             builder = SslContextBuilder.forServer(kmf).clientAuth(this.require_client_auth ? ClientAuth.REQUIRE :
                                                                   ClientAuth.NONE);
         }
         else
         {
+            KeyManagerFactory kmf = buildOutboundKeyManagerFactory();
             builder = SslContextBuilder.forClient().keyManager(kmf);
         }
 
@@ -263,4 +264,12 @@ protected SslProvider getSslProvider()
     abstract protected KeyManagerFactory buildKeyManagerFactory() throws SSLException;
 
     abstract protected TrustManagerFactory buildTrustManagerFactory() throws SSLException;
+
+    /**
+     * Create a {@code KeyManagerFactory} for outbound connections.
+     * It provides a seperate keystore for internode mTLS outbound connections.
+     * @return {@code KeyManagerFactory}
+     * @throws SSLException
+     */
+    abstract protected KeyManagerFactory buildOutboundKeyManagerFactory() throws SSLException;
 }

src/java/org/apache/cassandra/security/DisableSslContextFactory.java

@@ -36,12 +36,24 @@ protected TrustManagerFactory buildTrustManagerFactory() throws SSLException
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    protected KeyManagerFactory buildOutboundKeyManagerFactory() throws SSLException
+    {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public boolean hasKeystore()
     {
         return false;
     }
 
+    @Override
+    public boolean hasOutboundKeystore()
+    {
+        return false;
+    }
+
     @Override
     public void initHotReloading() throws SSLException
     {

src/java/org/apache/cassandra/security/FileBasedSslContextFactory.java

@@ -33,7 +33,7 @@
 import javax.net.ssl.SSLException;
 import javax.net.ssl.TrustManagerFactory;
 
-import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -47,38 +47,32 @@
  * {@code CAUTION:} While this is a useful abstraction, please be careful if you need to modify this class
  * given possible custom implementations out there!
  */
-abstract public class FileBasedSslContextFactory extends AbstractSslContextFactory
+public abstract class FileBasedSslContextFactory extends AbstractSslContextFactory
 {
     private static final Logger logger = LoggerFactory.getLogger(FileBasedSslContextFactory.class);
-
-    @VisibleForTesting
-    protected volatile boolean checkedExpiry = false;
+    protected FileBasedStoreContext keystoreContext;
+    protected FileBasedStoreContext outboundKeystoreContext;
+    protected FileBasedStoreContext trustStoreContext;
 
     /**
      * List of files that trigger hot reloading of SSL certificates
      */
     protected volatile List<HotReloadableFile> hotReloadableFiles = new ArrayList<>();
 
-    protected String keystore;
-    protected String keystore_password;
-    protected String truststore;
-    protected String truststore_password;
-
     public FileBasedSslContextFactory()
     {
-        keystore = "conf/.keystore";
-        keystore_password = "cassandra";
-        truststore = "conf/.truststore";
-        truststore_password = "cassandra";
+        keystoreContext = new FileBasedStoreContext("conf/.keystore", "cassandra");
+        outboundKeystoreContext = new FileBasedStoreContext("conf/.keystore", "cassandra");
+        trustStoreContext = new FileBasedStoreContext("conf/.truststore", "cassandra");
     }
 
     public FileBasedSslContextFactory(Map<String, Object> parameters)
     {
         super(parameters);
-        keystore = getString("keystore");
-        keystore_password = getString("keystore_password");
-        truststore = getString("truststore");
-        truststore_password = getString("truststore_password");
+        keystoreContext = new FileBasedStoreContext(getString("keystore"), getString("keystore_password"));
+        outboundKeystoreContext = new FileBasedStoreContext(StringUtils.defaultString(getString("outbound_keystore"), keystoreContext.filePath),
+                                                            StringUtils.defaultString(getString("outbound_keystore_password"), keystoreContext.password));
+        trustStoreContext = new FileBasedStoreContext(getString("truststore"), getString("truststore_password"));
     }
 
     @Override
@@ -90,30 +84,41 @@ public boolean shouldReload()
     @Override
     public boolean hasKeystore()
     {
-        return keystore != null && new File(keystore).exists();
+        return keystoreContext.hasKeystore();
+    }
+
+    @Override
+    public boolean hasOutboundKeystore()
+    {
+        return outboundKeystoreContext.hasKeystore();
     }
 
     private boolean hasTruststore()
     {
-        return truststore != null && new File(truststore).exists();
+        return trustStoreContext.filePath != null && new File(trustStoreContext.filePath).exists();
     }
 
     @Override
     public synchronized void initHotReloading()
     {
         boolean hasKeystore = hasKeystore();
+        boolean hasOutboundKeystore = hasOutboundKeystore();
         boolean hasTruststore = hasTruststore();
 
-        if (hasKeystore || hasTruststore)
+        if (hasKeystore || hasOutboundKeystore || hasTruststore)
         {
             List<HotReloadableFile> fileList = new ArrayList<>();
             if (hasKeystore)
             {
-                fileList.add(new HotReloadableFile(keystore));
+                fileList.add(new HotReloadableFile(keystoreContext.filePath));
+            }
+            if (hasOutboundKeystore)
+            {
+                fileList.add(new HotReloadableFile(outboundKeystoreContext.filePath));
             }
             if (hasTruststore)
             {
-                fileList.add(new HotReloadableFile(truststore));
+                fileList.add(new HotReloadableFile(trustStoreContext.filePath));
             }
             hotReloadableFiles = fileList;
         }
@@ -129,25 +134,13 @@ public synchronized void initHotReloading()
     @Override
     protected KeyManagerFactory buildKeyManagerFactory() throws SSLException
     {
+        return getKeyManagerFactory(keystoreContext);
+    }
 
-        try (InputStream ksf = Files.newInputStream(Paths.get(keystore)))
-        {
-            final String algorithm = this.algorithm == null ? KeyManagerFactory.getDefaultAlgorithm() : this.algorithm;
-            KeyManagerFactory kmf = KeyManagerFactory.getInstance(algorithm);
-            KeyStore ks = KeyStore.getInstance(store_type);
-            ks.load(ksf, keystore_password.toCharArray());
-            if (!checkedExpiry)
-            {
-                checkExpiredCerts(ks);
-                checkedExpiry = true;
-            }
-            kmf.init(ks, keystore_password.toCharArray());
-            return kmf;
-        }
-        catch (Exception e)
-        {
-            throw new SSLException("failed to build key manager store for secure connections", e);
-        }
+    @Override
+    protected KeyManagerFactory buildOutboundKeyManagerFactory() throws SSLException
+    {
+        return getKeyManagerFactory(outboundKeystoreContext);
     }
 
     /**
@@ -159,12 +152,12 @@ protected KeyManagerFactory buildKeyManagerFactory() throws SSLException
     @Override
     protected TrustManagerFactory buildTrustManagerFactory() throws SSLException
     {
-        try (InputStream tsf = Files.newInputStream(Paths.get(truststore)))
+        try (InputStream tsf = Files.newInputStream(Paths.get(trustStoreContext.filePath)))
         {
             final String algorithm = this.algorithm == null ? TrustManagerFactory.getDefaultAlgorithm() : this.algorithm;
             TrustManagerFactory tmf = TrustManagerFactory.getInstance(algorithm);
             KeyStore ts = KeyStore.getInstance(store_type);
-            ts.load(tsf, truststore_password.toCharArray());
+            ts.load(tsf, trustStoreContext.password.toCharArray());
             tmf.init(ts);
             return tmf;
         }
@@ -174,6 +167,29 @@ protected TrustManagerFactory buildTrustManagerFactory() throws SSLException
         }
     }
 
+    private KeyManagerFactory getKeyManagerFactory(final FileBasedStoreContext context) throws SSLException
+    {
+        try (InputStream ksf = Files.newInputStream(Paths.get(context.filePath)))
+        {
+            final String algorithm = this.algorithm == null ? KeyManagerFactory.getDefaultAlgorithm() : this.algorithm;
+            KeyManagerFactory kmf = KeyManagerFactory.getInstance(algorithm);
+            KeyStore ks = KeyStore.getInstance(store_type);
+            ks.load(ksf, context.password.toCharArray());
+
+            if (!context.checkedExpiry)
+            {
+                checkExpiredCerts(ks);
+                context.checkedExpiry = true;
+            }
+            kmf.init(ks, context.password.toCharArray());
+            return kmf;
+        }
+        catch (Exception e)
+        {
+            throw new SSLException("failed to build key manager store for secure connections", e);
+        }
+    }
+
     protected boolean checkExpiredCerts(KeyStore ks) throws KeyStoreException
     {
         boolean hasExpiredCerts = false;
@@ -225,4 +241,27 @@ public String toString()
                    '}';
         }
     }
+
+    protected static class FileBasedStoreContext
+    {
+        public volatile boolean checkedExpiry = false;
+        public String filePath;
+        public String password;
+
+        public FileBasedStoreContext(String keystore, String keystorePassword)
+        {
+            this.filePath = keystore;
+            this.password = keystorePassword;
+        }
+
+        protected boolean hasKeystore()
+        {
+            return filePath != null && new File(filePath).exists();
+        }
+
+        protected boolean passwordMatchesIfPresent(String keyPassword)
+        {
+            return StringUtils.isEmpty(password) || keyPassword.equals(password);
+        }
+    }
 }

src/java/org/apache/cassandra/security/ISslContextFactory.java

@@ -99,6 +99,16 @@ default boolean hasKeystore()
         return true;
     }
 
+    /**
+     * Returns if this factory uses outbound keystore.
+     *
+     * @return {@code true} by default unless the implementation overrides this
+     */
+    default boolean hasOutboundKeystore()
+    {
+        return false;
+    }
+
     /**
      * Returns the prepared list of accepted protocols.
      *