HCD-82 Enables auth* classes configuration

CHANGES.txt

@@ -1,6 +1,7 @@
 Future version (tbd)
  * Require only MODIFY permission on base when updating table with MV (STAR-564)
 Merged from 5.0:
+ * Use ParameterizedClass for all auth-related implementations (CASSANDRA-19946 and partially CASSANDRA-18554)
  * Enables IAuthenticator's to return own AuthenticateMessage (CASSANDRA-19984)
  * Disable chronicle analytics (CASSANDRA-19656)
  * Remove mocking in InternalNodeProbe spying on StorageServiceMBean (CASSANDRA-18152)

conf/cassandra.yaml

@@ -127,6 +127,10 @@ batchlog_replay_throttle_in_kb: 1024
 # Authentication backend, implementing IAuthenticator; used to identify users
 # Out of the box, Cassandra provides org.apache.cassandra.auth.{AllowAllAuthenticator,
 # PasswordAuthenticator}.
+# Optional parameters can be specified in the form of:
+#   parameters:
+#    param_key1: param_value1
+#    ...
 #
 # - AllowAllAuthenticator performs no checks - set it to disable authentication.
 # - PasswordAuthenticator relies on username/password pairs to authenticate
@@ -138,6 +142,10 @@ authenticator: AllowAllAuthenticator
 # Authorization backend, implementing IAuthorizer; used to limit access/provide permissions
 # Out of the box, Cassandra provides org.apache.cassandra.auth.{AllowAllAuthorizer,
 # CassandraAuthorizer}.
+# Optional parameters can be specified in the form of:
+#   parameters:
+#    param_key1: param_value1
+#    ...
 #
 # - AllowAllAuthorizer allows any action to any user - set it to disable authorization.
 # - CassandraAuthorizer stores permissions in system_auth.role_permissions table. Please
@@ -150,6 +158,10 @@ authorizer: AllowAllAuthorizer
 # which stores role information in the system_auth keyspace. Most functions of the
 # IRoleManager require an authenticated login, so unless the configured IAuthenticator
 # actually implements authentication, most of this functionality will be unavailable.
+# Optional parameters can be specified in the form of:
+#   parameters:
+#    param_key1: param_value1
+#    ...
 #
 # - CassandraRoleManager stores role data in the system_auth keyspace. Please
 #   increase system_auth keyspace replication factor if you use this role manager.
@@ -159,6 +171,10 @@ role_manager: CassandraRoleManager
 # access to certain DCs
 # Out of the box, Cassandra provides org.apache.cassandra.auth.{AllowAllNetworkAuthorizer,
 # CassandraNetworkAuthorizer}.
+# Optional parameters can be specified in the form of:
+#   parameters:
+#    param_key1: param_value1
+#    ...
 #
 # - AllowAllNetworkAuthorizer allows access to any DC to any user - set it to disable authorization.
 # - CassandraNetworkAuthorizer stores permissions in system_auth.network_permissions table. Please

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

@@ -18,13 +18,15 @@
 
 package org.apache.cassandra.auth;
 
+import java.util.List;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.ParameterizedClass;
 import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.utils.FBUtilities;
 
 /**
  * Only purpose is to Initialize authentication/authorization via {@link #applyAuth()}.
@@ -46,11 +48,10 @@ public static void applyAuth()
 
         Config conf = DatabaseDescriptor.getRawConfig();
 
-        IAuthenticator authenticator = new AllowAllAuthenticator();
 
-        /* Authentication, authorization and role management backend, implementing IAuthenticator, IAuthorizer & IRoleMapper*/
-        if (conf.authenticator != null)
-            authenticator = FBUtilities.newAuthenticator(conf.authenticator);
+        /* Authentication, authorization and role management backend, implementing IAuthenticator, I*Authorizer & IRoleManager */
+
+        IAuthenticator authenticator = authInstantiate(conf.authenticator, AllowAllAuthenticator.class);
 
         // the configuration options regarding credentials caching are only guaranteed to
         // work with PasswordAuthenticator, so log a message if some other authenticator
@@ -69,40 +70,39 @@ public static void applyAuth()
 
         // authorizer
 
-        IAuthorizer authorizer = new AllowAllAuthorizer();
-
-        if (conf.authorizer != null)
-            authorizer = FBUtilities.newAuthorizer(conf.authorizer);
+        IAuthorizer authorizer = authInstantiate(conf.authorizer, AllowAllAuthorizer.class);
 
         if (!authenticator.requireAuthentication() && authorizer.requireAuthorization())
-            throw new ConfigurationException(conf.authenticator + " can't be used with " + conf.authorizer, false);
+        {
+            throw new ConfigurationException(authorizer.getClass().getName() + " has authorization enabled which requires " +
+                                             authenticator.getClass().getName() + " to enable authentication", false);
+        }
 
         DatabaseDescriptor.setAuthorizer(authorizer);
 
         // role manager
 
-        IRoleManager roleManager;
-        if (conf.role_manager != null)
-            roleManager = FBUtilities.newRoleManager(conf.role_manager);
-        else
-            roleManager = new CassandraRoleManager();
+        IRoleManager roleManager = authInstantiate(conf.role_manager, CassandraRoleManager.class);
 
         if (authenticator instanceof PasswordAuthenticator && !(roleManager instanceof CassandraRoleManager))
-            throw new ConfigurationException("CassandraRoleManager must be used with PasswordAuthenticator", false);
+            throw new ConfigurationException(authenticator.getClass().getName() + " requires CassandraRoleManager", false);
 
         DatabaseDescriptor.setRoleManager(roleManager);
 
         // authenticator
 
-        if (conf.internode_authenticator != null)
-            DatabaseDescriptor.setInternodeAuthenticator(FBUtilities.construct(conf.internode_authenticator, "internode_authenticator"));
+        IInternodeAuthenticator internodeAuthenticator = authInstantiate(conf.internode_authenticator,
+                                                                         AllowAllInternodeAuthenticator.class);
+        DatabaseDescriptor.setInternodeAuthenticator(internodeAuthenticator);
 
         // network authorizer
-        INetworkAuthorizer networkAuthorizer = FBUtilities.newNetworkAuthorizer(conf.network_authorizer);
+
+        INetworkAuthorizer networkAuthorizer = authInstantiate(conf.network_authorizer, AllowAllNetworkAuthorizer.class);
         DatabaseDescriptor.setNetworkAuthorizer(networkAuthorizer);
+
         if (networkAuthorizer.requireAuthorization() && !authenticator.requireAuthentication())
         {
-            throw new ConfigurationException(conf.network_authorizer + " can't be used with " + conf.authenticator, false);
+            throw new ConfigurationException(conf.network_authorizer + " can't be used with " + conf.authenticator.class_name, false);
         }
 
         // Validate at last to have authenticator, authorizer, role-manager and internode-auth setup
@@ -114,4 +114,21 @@ public static void applyAuth()
         networkAuthorizer.validateConfiguration();
         DatabaseDescriptor.getInternodeAuthenticator().validateConfiguration();
     }
+
+    private static <T> T authInstantiate(ParameterizedClass authCls, Class<T> defaultCls) {
+        if (authCls != null && authCls.class_name != null)
+        {
+            String authPackage = AuthConfig.class.getPackage().getName();
+            return ParameterizedClass.newInstance(authCls, List.of("", authPackage));
+        }
+
+        try
+        {
+            return defaultCls.newInstance();
+        }
+        catch (InstantiationException | IllegalAccessException  e)
+        {
+            throw new ConfigurationException("Failed to instantiate " + defaultCls.getName(), e);
+        }
+    }
 }

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

@@ -23,6 +23,7 @@
 import java.util.Map;
 import java.util.Set;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Lists;
 import org.slf4j.Logger;
@@ -171,7 +172,8 @@ private static SelectStatement prepare(String query)
         return (SelectStatement) QueryProcessor.getStatement(query, ClientState.forInternalCalls());
     }
 
-    private class PlainTextSaslAuthenticator implements SaslNegotiator
+    @VisibleForTesting
+    class PlainTextSaslAuthenticator implements SaslNegotiator
     {
         private boolean complete = false;
         private String username;

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

@@ -54,10 +54,10 @@ public class Config
     public static final String PROPERTY_PREFIX = "cassandra.";
 
     public String cluster_name = "Test Cluster";
-    public String authenticator;
-    public String authorizer;
-    public String role_manager;
-    public String network_authorizer;
+    public ParameterizedClass authenticator;
+    public ParameterizedClass authorizer;
+    public ParameterizedClass role_manager;
+    public ParameterizedClass network_authorizer;
     public volatile int permissions_validity_in_ms = 2000;
     public volatile int permissions_cache_max_entries = 1000;
     public volatile int permissions_update_interval_in_ms = -1;
@@ -158,7 +158,7 @@ public class Config
     public boolean listen_interface_prefer_ipv6 = false;
     public String broadcast_address;
     public boolean listen_on_broadcast_address = false;
-    public String internode_authenticator;
+    public ParameterizedClass internode_authenticator;
 
     /*
      * RPC address and interface refer to the address/interface used for the native protocol used to communicate with

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

@@ -17,11 +17,19 @@
  */
 package org.apache.cassandra.config;
 
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import com.google.common.base.Objects;
 
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.utils.IntegerInterval;
+
 public class ParameterizedClass
 {
     public static final String CLASS_NAME = "class_name";
@@ -35,6 +43,12 @@ public ParameterizedClass()
         // for snakeyaml
     }
 
+    public ParameterizedClass(String class_name)
+    {
+        this.class_name = class_name;
+        this.parameters = Collections.emptyMap();
+    }
+
     public ParameterizedClass(String class_name, Map<String, String> parameters)
     {
         this.class_name = class_name;
@@ -48,6 +62,67 @@ public ParameterizedClass(Map<String, ?> p)
              p.containsKey(PARAMETERS) ? (Map<String, String>)((List<?>)p.get(PARAMETERS)).get(0) : null);
     }
 
+    static public <K> K newInstance(ParameterizedClass parameterizedClass, List<String> searchPackages)
+    {
+        Class<?> providerClass = null;
+        if (searchPackages == null || searchPackages.isEmpty())
+            searchPackages = Collections.singletonList("");
+        for (String searchPackage : searchPackages)
+        {
+            try
+            {
+                if (!searchPackage.isEmpty() && !searchPackage.endsWith("."))
+                    searchPackage = searchPackage + '.';
+                String name = searchPackage + parameterizedClass.class_name;
+                providerClass = Class.forName(name);
+            }
+            catch (ClassNotFoundException e)
+            {
+                //no-op
+            }
+        }
+
+        if (providerClass == null)
+        {
+            String error = "Unable to find class " + parameterizedClass.class_name + " in packages [" +
+                           searchPackages.stream().map(p -> '"' + p + '"').collect(Collectors.joining(",")) + ']';
+            throw new ConfigurationException(error);
+        }
+
+        try
+        {
+            Constructor<?>[] declaredConstructors = providerClass.getDeclaredConstructors();
+
+            Constructor mapConstructor = Arrays.stream(declaredConstructors)
+                                               .filter(c -> c.getParameterTypes().length == 1 && c.getParameterTypes()[0].equals(Map.class))
+                                               .findFirst().orElse(null);
+            if (mapConstructor != null)
+                return (K) mapConstructor.newInstance(parameterizedClass.parameters);
+
+            // Falls-back to no-arg constructor if no parameters are present
+            if (parameterizedClass.parameters == null || parameterizedClass.parameters.isEmpty())
+            {
+                Constructor emptyConstructor = Arrays.stream(declaredConstructors)
+                                                     .filter(c -> c.getParameterTypes().length == 0)
+                                                     .findFirst().orElse(null);
+                if (emptyConstructor != null)
+                    return (K) emptyConstructor.newInstance();
+            }
+
+            throw new ConfigurationException("No valid constructor found for class " + parameterizedClass.class_name);
+        }
+        catch (IllegalAccessException|InstantiationException|ExceptionInInitializerError e)
+        {
+            throw new ConfigurationException("Unable to instantiate parameterized class " + parameterizedClass.class_name, e);
+        }
+        catch (InvocationTargetException e)
+        {
+            Throwable cause = e.getCause();
+            String error = "Failed to instantiate class " + parameterizedClass.class_name + ": " + cause.getMessage();
+            throw new ConfigurationException(error, cause);
+        }
+    }
+
     @Override
     public boolean equals(Object that)
     {

src/java/org/apache/cassandra/utils/FBUtilities.java

@@ -72,11 +72,6 @@
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.cassandra.audit.IAuditLogger;
-import org.apache.cassandra.auth.AllowAllNetworkAuthorizer;
-import org.apache.cassandra.auth.IAuthenticator;
-import org.apache.cassandra.auth.IAuthorizer;
-import org.apache.cassandra.auth.INetworkAuthorizer;
-import org.apache.cassandra.auth.IRoleManager;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.SerializationHeader;
@@ -712,40 +707,6 @@ static IPartitioner newPartitioner(String partitionerClassName, Optional<Abstrac
         return FBUtilities.instanceOrConstruct(partitionerClassName, "partitioner");
     }
 
-    public static IAuthorizer newAuthorizer(String className) throws ConfigurationException
-    {
-        if (!className.contains("."))
-            className = "org.apache.cassandra.auth." + className;
-        return FBUtilities.construct(className, "authorizer");
-    }
-
-    public static IAuthenticator newAuthenticator(String className) throws ConfigurationException
-    {
-        if (!className.contains("."))
-            className = "org.apache.cassandra.auth." + className;
-        return FBUtilities.construct(className, "authenticator");
-    }
-
-    public static IRoleManager newRoleManager(String className) throws ConfigurationException
-    {
-        if (!className.contains("."))
-            className = "org.apache.cassandra.auth." + className;
-        return FBUtilities.construct(className, "role manager");
-    }
-
-    public static INetworkAuthorizer newNetworkAuthorizer(String className)
-    {
-        if (className == null)
-        {
-            return new AllowAllNetworkAuthorizer();
-        }
-        if (!className.contains("."))
-        {
-            className = "org.apache.cassandra.auth." + className;
-        }
-        return FBUtilities.construct(className, "network authorizer");
-    }
-
     public static IAuditLogger newAuditLogger(String className, Map<String, String> parameters) throws ConfigurationException
     {
         if (!className.contains("."))

test/unit/org/apache/cassandra/audit/AuditLoggerAuthTest.java

@@ -34,6 +34,9 @@
 import com.datastax.driver.core.exceptions.SyntaxError;
 import com.datastax.driver.core.exceptions.UnauthorizedException;
 import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.auth.CassandraAuthorizer;
+import org.apache.cassandra.auth.CassandraRoleManager;
+import org.apache.cassandra.auth.PasswordAuthenticator;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.OverrideConfigurationLoader;
 import org.apache.cassandra.config.ParameterizedClass;
@@ -67,11 +70,11 @@ public class AuditLoggerAuthTest
     public static void setup() throws Exception
     {
         OverrideConfigurationLoader.override((config) -> {
-            config.authenticator = "PasswordAuthenticator";
-            config.role_manager = "CassandraRoleManager";
-            config.authorizer = "CassandraAuthorizer";
+            config.authenticator = new ParameterizedClass(PasswordAuthenticator.class.getName());
+            config.role_manager = new ParameterizedClass(CassandraRoleManager.class.getName());
+            config.authorizer = new ParameterizedClass(CassandraAuthorizer.class.getName());
             config.audit_logging_options.enabled = true;
-            config.audit_logging_options.logger = new ParameterizedClass("InMemoryAuditLogger", null);
+            config.audit_logging_options.logger = new ParameterizedClass(InMemoryAuditLogger.class.getName(), null);
         });
 
         System.setProperty("cassandra.superuser_setup_delay_ms", "0");