added eligibility checks framework

Author: smiklosovic

README.adoc

@@ -89,6 +89,21 @@ The content of the configuration file is as follows:
 
 |load_ldap_service
 |defaults to false, if it is true, SPI mechanism will look on class path to load custom implementation of `LDAPUserRetriever`.
+
+|eligibility_class_name
+|defaults to `NoOpLoginEligibilityCheck`
+
+|eligibility_cassandra_keyspace
+|defaults to `login_eligibility`
+
+|eligibility_cassandra_table
+|defaults to `login_eligibility`
+
+|eligibility_cassandra_user_column
+|defaults to `user`
+
+|eligibility_cassandra_access_column
+|defaults to `has_access`
 |===
 
 
@@ -158,6 +173,70 @@ may have a different search filter based on your need, a lot of people use e.g.
 If you try to log in with `cqlsh -u cn=myuserinldap`, there will be no replacement done and this will be
 used as a search filter instead.
 
+## Login eligibility checks
+
+There is an additional mechanims implemented for login eligibility resoultion after user is authenticated
+in LDAP. If user is authenticated against LDAP and he is not able to log in (e.g. submitted password is wrong),
+this check is bypassed. However, even if a user in LDAP is able to log in, Cassandra database administrators
+can use additional checks to decide if a user who passed LDAP authentication procedure is indeed able to log in or not.
+
+The login eligibility check implementation is driven by the configuration property `eligibility_class_name` which
+defaults to the no-op implementation which means that login to LDAP will make such user eligible to log in to Cassandra
+without any further restrictions / checks.
+
+There is a default Cassandra implementation provided which might be used by setting `eligibility_class_name` to
+`com.instaclustr.cassandra.ldap.auth.Cassandra311LoginEligibilityCheck` for C* 3.11. If you are running Cassandra 4.0, please
+set this property to `com.instaclustr.cassandra.ldap.auth.Cassandra40LoginEligibilityCheck`.
+IF you use 3.0 or 2.2, use `CassandraLoginEligibilityCheck`.
+
+Before using this feature (when you are using Cassandra check), respective keyspace
+and table need to be created which will capture eligibility data. The default script
+is located in `conf/eligibility_check.cql`. Please keep in mind that you might
+alter this keyspace to e.g. reflect your replication strategy requirements - this duty is
+left to Cassandra operator.
+
+Operator is responsible for the population of this table, plugin just reads from this table
+and it expects that such record is found. For example, let's say that operator inserted this record:
+
+----
+cqlsh> INSERT INTO login_eligibility.login_eligibility (user , has_access ) VALUES ( 'cn=stefan,dc=example,dc=org', true) USING TTL 60;
+cqlsh> select user, has_access, ttl(has_access) from login_eligibility.login_eligibility where user = 'stefan';
+
+ user                        | has_access | ttl(has_access)
+-----------------------------+------------+-----------------
+ cn=stefan,dc=example,dc=org |       True |              55
+
+(1 rows)
+----
+
+The default Cassandra check implementation will do this check upon login:
+
+----
+clqsh> select user, has_access from login_eligibility.login_eligibility where user = 'cn=stefan,dc=example,dc=org';
+----
+
+Then the plugin looks into `has_access` and it has to be `true`.
+
+TTL is optional, here it is used with advantage that a user is eligible to be logged in
+just for 1 minute. There might be e.g. some company policy to enable a user to log in
+during 2 days, for example, which would mimic this policy. After 2 days, such record
+is not present in database anymore which renders a user to be unable to log in.
+
+If you want to use custom eligibility check implementation, you need to firstly implement the interface
+`com.instaclustr.cassandra.ldap.auth.LoginEligibilityCheck`, then you need to
+create a JAR with this class and put it on Cassandra's class path. In your JAR,
+you need to specify your implemenatation in `src/main/resources/META-INF/services` where you need
+to put a file with name of FQCN of interface and its content will be FQCN of your implementation which
+implements this interface.
+
+If you do not want to use SPI mechanism mentioned above, you still have to put your
+implementation on the class path but you have to specify your implementation
+in `eligibility_class_name` configuration property.
+
+If you want to use different configuration properties for your custom implementation, you have them
+available in interfaces' method `init` where they are passed from plugin internals upon
+its setup.
+
 ## How it Works
 
 LDAPAuthenticator currently supports plain text authorization requests only in the form of a username and password.

base/src/main/java/com/instaclustr/cassandra/ldap/auth/BaseCassandraLoginEligibilityCheck.java

@@ -0,0 +1,84 @@
+package com.instaclustr.cassandra.ldap.auth;
+
+import java.util.Properties;
+
+import com.instaclustr.cassandra.ldap.User;
+import com.instaclustr.cassandra.ldap.conf.LdapAuthenticatorConfiguration;
+import org.apache.cassandra.serializers.BooleanSerializer;
+import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.transport.messages.ResultMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BaseCassandraLoginEligibilityCheck implements LoginEligibilityCheck
+{
+    private static final Logger logger = LoggerFactory.getLogger(BaseCassandraLoginEligibilityCheck.class);
+
+    private static final String BASE_SELECT_USER_STATEMENT_TEMPLATE = "select %s from %s.%s where %s = ?";
+
+    protected ClientState clientState;
+    protected Properties configProperties;
+    protected String selectStatement;
+
+    @Override
+    public void init(final ClientState clientState, final Properties configProperties)
+    {
+        this.clientState = clientState;
+        this.configProperties = configProperties;
+
+        this.selectStatement = String.format(BASE_SELECT_USER_STATEMENT_TEMPLATE,
+                                             configProperties.getProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_ACCESS_COLUMN),
+                                             configProperties.getProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_KEYSPACE),
+                                             configProperties.getProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_TABLE),
+                                             configProperties.getProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_USER_COLUMN));
+
+    }
+
+    protected abstract ResultMessage.Rows getRows(final String loginName);
+
+    @Override
+    public boolean isEligibleToLogin(final User user, final String loginName)
+    {
+
+        // all non-ldap users are free to log in just fine
+        if (user.getLdapDN() == null)
+        {
+            return true;
+        }
+
+        assert clientState != null;
+
+        final ResultMessage.Rows rows = getRows(loginName);
+
+        final boolean noResults = rows.result.isEmpty();
+
+        if (noResults)
+        {
+            logger.info(String.format("User with login name '%s' is not eligible to be logged in!", loginName));
+            return false;
+        }
+        else
+        {
+            if (rows.result.size() != 1)
+            {
+                throw new IllegalStateException("There was more than one record returned from eligibility check select query!");
+            }
+
+            if (rows.result.rows.get(0).size() != 1)
+            {
+                throw new IllegalStateException("There was more than one column returned from eligibility check select query!");
+            }
+
+            if (BooleanSerializer.instance.deserialize(rows.result.rows.get(0).get(0)))
+            {
+                logger.info(String.format("User with login name '%s' is eligible to be logged in!", loginName));
+                return true;
+            }
+            else
+            {
+                logger.info(String.format("User with login name '%s' is not eligible to be logged in!", loginName));
+                return false;
+            }
+        }
+    }
+}

base/src/main/java/com/instaclustr/cassandra/ldap/auth/CassandraLoginEligibilityCheck.java

@@ -0,0 +1,20 @@
+package com.instaclustr.cassandra.ldap.auth;
+
+import static java.util.Collections.singletonList;
+
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.statements.SelectStatement;
+import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.messages.ResultMessage.Rows;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class CassandraLoginEligibilityCheck extends BaseCassandraLoginEligibilityCheck
+{
+    @Override
+    protected Rows getRows(final String loginName)
+    {
+        final SelectStatement selStmt = (SelectStatement) QueryProcessor.getStatement(selectStatement, clientState).statement;
+        return selStmt.execute(new QueryState(clientState), QueryOptions.forInternalCalls(singletonList(ByteBufferUtil.bytes(loginName))));
+    }
+}

base/src/main/java/com/instaclustr/cassandra/ldap/auth/LoginEligibilityCheck.java

@@ -0,0 +1,15 @@
+package com.instaclustr.cassandra.ldap.auth;
+
+import java.util.Properties;
+
+import com.instaclustr.cassandra.ldap.User;
+import org.apache.cassandra.service.ClientState;
+
+public interface LoginEligibilityCheck
+{
+
+    void init(final ClientState clientState, final Properties configProperties);
+
+    boolean isEligibleToLogin(final User user, final String loginName);
+
+}

base/src/main/java/com/instaclustr/cassandra/ldap/auth/NoOpLoginEligibilityCheck.java

@@ -0,0 +1,22 @@
+package com.instaclustr.cassandra.ldap.auth;
+
+import java.util.Properties;
+
+import com.instaclustr.cassandra.ldap.User;
+import org.apache.cassandra.service.ClientState;
+
+public final class NoOpLoginEligibilityCheck implements LoginEligibilityCheck
+{
+
+    @Override
+    public void init(final ClientState clientState, final Properties properties)
+    {
+
+    }
+
+    @Override
+    public boolean isEligibleToLogin(final User user, final String loginName)
+    {
+        return true;
+    }
+}

base/src/main/java/com/instaclustr/cassandra/ldap/conf/LdapAuthenticatorConfiguration.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.util.Properties;
 
+import com.instaclustr.cassandra.ldap.auth.NoOpLoginEligibilityCheck;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -60,6 +61,19 @@ public final class LdapAuthenticatorConfiguration
 
     public static final String CASSANDRA_LDAP_ADMIN_USER = "cassandra.ldap.admin.user";
 
+    public static final String ELIGIBILITY_CHECK_CLASS_NAME = "eligibility_class_name";
+    public static final String DEFAULT_ELIGIBILITY_CHECK_CLASS_NAME = NoOpLoginEligibilityCheck.class.getCanonicalName();
+
+    public static final String CASSANDRA_ELIGIBILITY_CHECK_KEYSPACE = "eligibility_cassandra_keyspace";
+    public static final String CASSANDRA_ELIGIBILITY_CHECK_TABLE = "eligibility_cassandra_table";
+    public static final String CASSANDRA_ELIGIBILITY_CHECK_USER_COLUMN = "eligibility_cassandra_user_column";
+    public static final String CASSANDRA_ELIGIBILITY_CHECK_ACCESS_COLUMN = "eligibility_cassandra_access_column";
+
+    public static final String DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_KEYSPACE = "login_eligibility";
+    public static final String DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_TABLE = "login_eligibility";
+    public static final String DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_USER_COLUMN = "user";
+    public static final String DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_ACCESS_COLUMN = "has_access";
+
     public static final String CONSISTENCY_FOR_ROLE = "consistency_for_role";
     public static final String DEFAULT_CONSISTENCY_FOR_ROLE = "LOCAL_ONE";
 
@@ -139,10 +153,22 @@ public Properties parseProperties() throws ConfigurationException
 
         properties.setProperty(FILTER_TEMPLATE, filterTemplate);
 
+        properties.setProperty(LdapAuthenticatorConfiguration.CONTEXT_FACTORY_PROP, properties.getProperty(CONTEXT_FACTORY_PROP, DEFAULT_CONTEXT_FACTORY));
+        properties.setProperty(LdapAuthenticatorConfiguration.LDAP_URI_PROP, properties.getProperty(LDAP_URI_PROP));
+
+        properties.setProperty(LdapAuthenticatorConfiguration.ELIGIBILITY_CHECK_CLASS_NAME, properties.getProperty(ELIGIBILITY_CHECK_CLASS_NAME, DEFAULT_ELIGIBILITY_CHECK_CLASS_NAME));
+
+        properties.setProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_KEYSPACE,
+                               properties.getProperty(CASSANDRA_ELIGIBILITY_CHECK_KEYSPACE, DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_KEYSPACE));
+
+        properties.setProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_TABLE,
+                               properties.getProperty(CASSANDRA_ELIGIBILITY_CHECK_TABLE, DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_TABLE));
 
+        properties.setProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_USER_COLUMN,
+                               properties.getProperty(CASSANDRA_ELIGIBILITY_CHECK_USER_COLUMN, DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_USER_COLUMN));
 
-        properties.put(LdapAuthenticatorConfiguration.CONTEXT_FACTORY_PROP, properties.getProperty(CONTEXT_FACTORY_PROP, DEFAULT_CONTEXT_FACTORY));
-        properties.put(LdapAuthenticatorConfiguration.LDAP_URI_PROP, properties.getProperty(LDAP_URI_PROP));
+        properties.setProperty(LdapAuthenticatorConfiguration.CASSANDRA_ELIGIBILITY_CHECK_ACCESS_COLUMN,
+                               properties.getProperty(CASSANDRA_ELIGIBILITY_CHECK_ACCESS_COLUMN, DEFAULT_CASSANDRA_ELIGIBILITY_CHECK_ACCESS_COLUMN));
 
         return properties;
     }

base/src/main/java/com/instaclustr/cassandra/ldap/utils/ServiceUtils.java

@@ -34,6 +34,20 @@ public class ServiceUtils
 
     private static final Logger logger = LoggerFactory.getLogger(ServiceUtils.class);
 
+    public static <T> T getServiceFromConfig(final Class<T> clazz,
+                                             final String classNameFromConfig)
+    {
+        Class<?> defaultImplClazz;
+
+        try {
+            defaultImplClazz = Class.forName(classNameFromConfig);
+        } catch (final ClassNotFoundException e) {
+            throw new IllegalStateException(format("Could not find class %s", classNameFromConfig));
+        }
+
+        return getService(clazz, (Class<T>) defaultImplClazz);
+    }
+
     public static <T> T getService(final Class<T> clazz, final Class<? extends T> defaultImplClazz)
     {
         final ServiceLoader<T> loader = ServiceLoader.load(clazz);

base/src/main/java/org/apache/cassandra/auth/LDAPAuthenticator.java

@@ -19,7 +19,9 @@
 
 import static com.instaclustr.cassandra.ldap.conf.LdapAuthenticatorConfiguration.CASSANDRA_AUTH_CACHE_ENABLED_PROP;
 import static com.instaclustr.cassandra.ldap.conf.LdapAuthenticatorConfiguration.CASSANDRA_LDAP_ADMIN_USER;
+import static com.instaclustr.cassandra.ldap.conf.LdapAuthenticatorConfiguration.ELIGIBILITY_CHECK_CLASS_NAME;
 import static com.instaclustr.cassandra.ldap.utils.ServiceUtils.getService;
+import static com.instaclustr.cassandra.ldap.utils.ServiceUtils.getServiceFromConfig;
 import static java.lang.Boolean.parseBoolean;
 import static java.lang.String.format;
 
@@ -29,6 +31,7 @@
 import com.google.common.util.concurrent.UncheckedExecutionException;
 import com.google.common.util.concurrent.Uninterruptibles;
 import com.instaclustr.cassandra.ldap.AbstractLDAPAuthenticator;
+import com.instaclustr.cassandra.ldap.auth.LoginEligibilityCheck;
 import com.instaclustr.cassandra.ldap.PlainTextSaslAuthenticator;
 import com.instaclustr.cassandra.ldap.User;
 import com.instaclustr.cassandra.ldap.auth.CassandraUserRetriever;
@@ -64,6 +67,7 @@ public class LDAPAuthenticator extends AbstractLDAPAuthenticator
     private static final Logger logger = LoggerFactory.getLogger(AbstractLDAPAuthenticator.class);
 
     protected CacheDelegate cacheDelegate;
+    protected LoginEligibilityCheck loginEligibilityCheck;
 
     public void setup()
     {
@@ -85,6 +89,9 @@ public void setup()
 
         cacheDelegate = getService(CacheDelegate.class, null);
 
+        loginEligibilityCheck = getServiceFromConfig(LoginEligibilityCheck.class, properties.getProperty(ELIGIBILITY_CHECK_CLASS_NAME));
+        loginEligibilityCheck.init(clientState, properties);
+
         final String adminRole = System.getProperty(CASSANDRA_LDAP_ADMIN_USER, "cassandra");
 
         while (true)
@@ -166,9 +173,16 @@ public AuthenticatedUser authenticate(String username, String password) throws A
 
                 final String loginName = cachedUser.getLdapDN() == null ? cachedUser.getUsername() : cachedUser.getLdapDN();
 
-                logger.debug("Going to log in with {}", loginName);
-
-                return new AuthenticatedUser(loginName);
+                if (loginEligibilityCheck.isEligibleToLogin(cachedUser, loginName))
+                {
+                    logger.debug("Going to log in with {}", loginName);
+                    return new AuthenticatedUser(loginName);
+                }
+                else
+                {
+                    throw new AuthenticationException(String.format("User %s is authenticated against LDAP fine but it is not able "
+                                                                        + "to log in based on an eligibility check.", loginName));
+                }
             }
         } catch (final UncheckedExecutionException ex)
         {

cassandra-3.11/src/main/java/com/instaclustr/cassandra/ldap/auth/Cassandra311LoginEligibilityCheck.java

@@ -0,0 +1,22 @@
+package com.instaclustr.cassandra.ldap.auth;
+
+import static java.util.Collections.singletonList;
+
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.statements.SelectStatement;
+import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.messages.ResultMessage.Rows;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class Cassandra311LoginEligibilityCheck extends BaseCassandraLoginEligibilityCheck
+{
+    @Override
+    protected Rows getRows(final String loginName)
+    {
+        final SelectStatement selStmt = (SelectStatement) QueryProcessor.getStatement(selectStatement, clientState).statement;
+        return selStmt.execute(new QueryState(clientState),
+                               QueryOptions.forInternalCalls(singletonList(ByteBufferUtil.bytes(loginName))),
+                               System.nanoTime());
+    }
+}