Support pluggable internode authentication

patch by Aleksey Yeschenko; reviewed by Brandon Williams for
CASSANDRA-5401

CHANGES.txt

@@ -12,6 +12,7 @@
  * cli: Quote ks and cf names in schema output when needed (CASSANDRA-5052)
  * Fix bad default for min/max timestamp in SSTableMetadata (CASSANDRA-5372)
  * Fix cf name extraction from manifest in Directories.migrateFile() (CASSANDRA-5242)
+ * Support pluggable internode authentication (CASSANDRA-5401)
 
 
 1.1.10

NEWS.txt

@@ -8,6 +8,19 @@ upgrade, just in case you need to roll back to the previous version.
 (Cassandra version X + 1 will always be able to read data files created
 by version X, but the inverse is not necessarily the case.)
 
+1.1.11
+======
+
+Upgrading
+---------
+    - Nothing specific to this release, but please see the previous instructions
+      if you are not upgrading from 1.1.10.
+
+Features
+--------
+    - Pluggable internode authentication.
+      See `internode_authenticator` setting in cassandra.yaml.
+
 1.1.10
 ======
 

conf/cassandra.yaml

@@ -275,6 +275,10 @@ listen_address: localhost
 # Leaving this blank will set it to the same value as listen_address
 # broadcast_address: 1.2.3.4
 
+# Internode authentication backend, implementing IInternodeAuthenticator;
+# used to allow/disallow connections from peer nodes.
+# internode_authenticator: org.apache.cassandra.auth.AllowAllInternodeAuthenticator
+
 # The address to bind the Thrift RPC service to -- clients connect
 # here. Unlike ListenAddress above, you *can* specify 0.0.0.0 here if
 # you want Thrift to listen on all interfaces.

src/java/org/apache/cassandra/auth/AllowAllInternodeAuthenticator.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cassandra.auth;
+
+import java.net.InetAddress;
+
+import org.apache.cassandra.config.ConfigurationException;
+
+public class AllowAllInternodeAuthenticator implements IInternodeAuthenticator
+{
+    public boolean authenticate(InetAddress remoteAddress, int remotePort)
+    {
+        return true;
+    }
+
+    public void validateConfiguration() throws ConfigurationException
+    {
+    }
+}

src/java/org/apache/cassandra/auth/IInternodeAuthenticator.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cassandra.auth;
+
+import java.net.InetAddress;
+
+import org.apache.cassandra.config.ConfigurationException;
+
+public interface IInternodeAuthenticator
+{
+    /**
+     * Decides whether or not a peer is allowed to connect to this node.
+     * If this method returns false, the socket will be immediately closed.
+     *
+     * @param remoteAddress ip address of the connecting node.
+     * @param remotePort port of the connecting node.
+     * @return true if the connection should be accepted, false otherwise.
+     */
+    boolean authenticate(InetAddress remoteAddress, int remotePort);
+
+    /**
+     * Validates configuration of IInternodeAuthenticator implementation (if configurable).
+     *
+     * @throws ConfigurationException when there is a configuration error.
+     */
+    void validateConfiguration() throws ConfigurationException;
+}

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

@@ -66,6 +66,7 @@ public class Config
     public Integer ssl_storage_port = 7001;
     public String listen_address;
     public String broadcast_address;
+    public String internode_authenticator;
 
     public String rpc_address;
     public Integer rpc_port = 9160;

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

@@ -63,6 +63,7 @@ public class DatabaseDescriptor
     private static InetAddress broadcastAddress;
     private static InetAddress rpcAddress;
     private static SeedProvider seedProvider;
+    private static IInternodeAuthenticator internodeAuthenticator;
 
     /* Hashing strategy Random or OPHF */
     private static IPartitioner partitioner;
@@ -201,14 +202,22 @@ public class DatabaseDescriptor
                 authenticator = FBUtilities.<IAuthenticator>construct(conf.authenticator, "authenticator");
             if (conf.authority != null)
                 authority = FBUtilities.<IAuthority>construct(conf.authority, "authority");
+
+            if (conf.internode_authenticator != null)
+                internodeAuthenticator = FBUtilities.construct(conf.internode_authenticator, "internode_authenticator");
+            else
+                internodeAuthenticator = new AllowAllInternodeAuthenticator();
+
             authenticator.validateConfiguration();
             authority.validateConfiguration();
+            internodeAuthenticator.validateConfiguration();
 
             /* Hashing strategy */
             if (conf.partitioner == null)
             {
                 throw new ConfigurationException("Missing directive: partitioner");
             }
+
             try
             {
                 partitioner = FBUtilities.newPartitioner(System.getProperty("cassandra.partitioner", conf.partitioner));
@@ -812,6 +821,11 @@ public class DatabaseDescriptor
         return broadcastAddress;
     }
 
+    public static IInternodeAuthenticator getInternodeAuthenticator()
+    {
+        return internodeAuthenticator;
+    }
+
     public static void setBroadcastAddress(InetAddress broadcastAdd)
     {
         broadcastAddress = broadcastAdd;

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

@@ -698,7 +698,10 @@ public final class MessagingService implements MessagingServiceMBean
                 try
                 {
                     Socket socket = server.accept();
-                    new IncomingTcpConnection(socket).start();
+                    if (authenticate(socket))
+                        new IncomingTcpConnection(socket).start();
+                    else
+                        socket.close();
                 }
                 catch (AsynchronousCloseException e)
                 {
@@ -717,6 +720,11 @@ public final class MessagingService implements MessagingServiceMBean
         {
             server.close();
         }
+
+        private boolean authenticate(Socket socket)
+        {
+            return DatabaseDescriptor.getInternodeAuthenticator().authenticate(socket.getInetAddress(), socket.getPort());
+        }
     }
 
     public Map<String, Integer> getCommandPendingTasks()