Make wrapped authorizer backend configurable

Closes Ericsson#104

CHANGES.md

@@ -1,6 +1,7 @@
 # Changes
 
 ## Version 2.1.0
+* Make wrapped authorizer backend configurable - #104
 * Add support for client port in log messages - #90
 * Add support for post-logging - #24
 * Add support for host address in log message - #28

conf/audit.yaml

@@ -19,10 +19,24 @@
 # directory.
 
 
+# The authorizer backend where the AuditAuthorizer delegate authorization requests.
+#
+# The value must represent a class name implementing the IAuthorizer interface. This can be any of the backends provided
+# out of the box by Cassandra, or your own custom implementation. Cassandra comes with the following authorizer
+# implementations:
+# - org.apache.cassandra.auth.AllowAllAuthorizer allows any action to any user - set it to disable authorization.
+# - org.apache.cassandra.auth.CassandraAuthorizer stores permissions in system_auth.role_permissions table.
+#
+# By default ecAudit will delegate to the CassandraAuthorizer.
+#
+wrapped_authorizer: org.apache.cassandra.auth.CassandraAuthorizer
+
+
 # Role/User whitelist at node level
 # This whitelist is only considered if the Java property 'ecaudit.filter_type' is set to YAML or YAML_AND_ROLE
 # When enabled, log entries will not be generated for queries performed by roles listed here.
 # Authentication attempts will still be logged.
+#
 #whitelist:
 #    - username1
 #    - username2
@@ -36,7 +50,8 @@
 #
 # post_logging -> One message will always be logged after the operation has completed. That message will indicate
 #                 whether operation SUCCEEDED or FAILED. This is how logging is implemented in C* 4.0.
-log_timing_strategy: post_logging
+#
+log_timing_strategy: pre_logging
 
 
 # Audit logger backend, implementing the AuditLogger interface
@@ -45,6 +60,8 @@ log_timing_strategy: post_logging
 # - com.ericsson.bss.cassandra.ecaudit.logger.Slf4jAuditLogger
 # - com.ericsson.bss.cassandra.ecaudit.logger.ChronicleAuditLogger
 #
+# By default ecAudit will use the Slf4jAuditLogger.
+#
 # Slf4jAuditLogger: Configure you logback.xml to specify location of log files and rotation policy. The following
 # optional parameters are accepted:
 # - log_format   - Format of the audit record sent to SLF4J. Fields can be configured with bash-style parameter

doc/cassandra_compatibility.md

@@ -82,14 +82,15 @@ The most notable being:
   When Cassandra 4.0 do separate statements, they will be prepended with a summary record indicating number of records in the batch.
   The ecAudit plug-in will not write a batch summary record.
 
-* ecAudit requires authentication and authorization backends to be enabled.
+* ecAudit requires an authentication backend to be enabled.
   Cassandra 4.0 have no such requirements.
   This is being addressed in [#77](https://github.com/Ericsson/ecaudit/issues/77).
 
 * ecAudit will log values for prepared statements.
   Audit logs in Cassandra 4.0 will not but this is being addressd in [CASSANDRA-14465](https://issues.apache.org/jira/browse/CASSANDRA-14465)
   FQL (Full Query Log) records in Cassandra 4.0 will include values for prepared statements.
 
+
 ### Audit Record Format
 
 Cassandra 4.0 is creating a fixed record format with some optional fields.

doc/setup.md

@@ -22,6 +22,16 @@ Read on below to learn how to tune the logger backend and manage audit whitelist
 
 In the [audit.yaml reference](audit_yaml_reference.md) you'll find more details about different options.
 
+
+### Wrapped Authorizer Backend
+
+The ecAudit plug-in must be installed as the ```authorizer``` in the ```cassandra.yaml``` in order to capture some operations.
+The actual authorization still has to be handled by another ```IAuthorizer``` implementation.
+Cassandra provides the ```AllowAllAuthorizer``` and the ```CassandraAuthorizer``` out of the box,
+where ecAudit will use the ```CassandraAuthorizer``` by default.
+With the ```wrapped_authorizer``` setting in the ```audit.yaml``` file it is possible to configure another ```IAuthorizer``` which may be your own custom implementation.
+
+
 ### Logger Backend
 
 Two different logger backends are supported out of the box:
@@ -32,11 +42,13 @@ Two different logger backends are supported out of the box:
 * Then there is the [Chronicle Logger](chronicle_logger.md) backend which has the best performance characteristics.
   This backend stores audit records in a binary format and is best suited when handling large volumes of records.
 
+
 ### Logger Timing
 
 Logger timing specifies *when* log entries should be written, **pre-logging** (default) and **post-logging** (C* 4.0 style) are available.
 You'll find more details in the [audit.yaml reference](audit_yaml_reference.md).
 
+
 ### Audit Whitelists
 
 By default ecAudit will create a record for each and every login attempt and CQL query.

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/AuditAdapter.java

@@ -30,6 +30,7 @@
 import com.ericsson.bss.cassandra.ecaudit.entry.PreparedAuditOperation;
 import com.ericsson.bss.cassandra.ecaudit.entry.factory.AuditEntryBuilderFactory;
 import com.ericsson.bss.cassandra.ecaudit.facade.Auditor;
+import com.ericsson.bss.cassandra.ecaudit.utils.Exceptions;
 import org.apache.cassandra.cql3.BatchQueryOptions;
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.QueryOptions;
@@ -40,8 +41,6 @@
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.MD5Digest;
 
-import static com.ericsson.bss.cassandra.ecaudit.auth.Exceptions.appendCause;
-
 /**
  * This class will be responsible for populating {@link AuditEntry} instance and passing that to {@link Auditor} instance
  */
@@ -203,12 +202,12 @@ public void auditAuth(String username, InetAddress clientIp, Status status, long
             }
             catch (RequestExecutionException e)
             {
-                throw appendCause(new AuthenticationException(e.toString()), e);
+                throw Exceptions.appendCause(new AuthenticationException(e.toString()), e);
             }
         }
     }
 
-    protected static final SimpleAuditOperation statusToAuthenticationOperation(Status status)
+    static SimpleAuditOperation statusToAuthenticationOperation(Status status)
     {
         return new SimpleAuditOperation("Authentication " + status.getDisplayName());
     }

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/auth/AuditAuthorizer.java

@@ -22,8 +22,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.ericsson.bss.cassandra.ecaudit.config.AuditConfig;
 import org.apache.cassandra.auth.AuthenticatedUser;
-import org.apache.cassandra.auth.CassandraAuthorizer;
 import org.apache.cassandra.auth.IAuthorizer;
 import org.apache.cassandra.auth.IResource;
 import org.apache.cassandra.auth.Permission;
@@ -33,9 +33,10 @@
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.RequestExecutionException;
 import org.apache.cassandra.exceptions.RequestValidationException;
+import org.apache.cassandra.utils.FBUtilities;
 
 /**
- * A decorator of the standard {@link CassandraAuthorizer} which always allow ALTER permission on {@link RoleResource}s.
+ * A decorator of a generic {@link IAuthorizer} which always allow ALTER permission on {@link RoleResource}s.
  *
  * This will allow one role to grant whitelist permission to another role by changing the OPTIONS attribute of the other role.
  * Other attributes such as PASSWORD may only be ALTERed if the permission actually have been assigned.
@@ -49,7 +50,14 @@ public class AuditAuthorizer implements IAuthorizer
 
     public AuditAuthorizer()
     {
-        this(new CassandraAuthorizer());
+        this(newWrappedAuthorizer(AuditConfig.getInstance()));
+    }
+
+    @VisibleForTesting
+    static IAuthorizer newWrappedAuthorizer(AuditConfig auditConfig)
+    {
+        String authorizerName = auditConfig.getWrappedAuthorizer();
+        return FBUtilities.newAuthorizer(authorizerName);
     }
 
     @VisibleForTesting

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/auth/WhitelistOptionParser.java

@@ -19,12 +19,11 @@
 
 import com.google.common.collect.ImmutableSet;
 
+import com.ericsson.bss.cassandra.ecaudit.utils.Exceptions;
 import org.apache.cassandra.auth.IResource;
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 
-import static com.ericsson.bss.cassandra.ecaudit.auth.Exceptions.appendCause;
-
 class WhitelistOptionParser
 {
     private static final String ALL_OPERATIONS = "ALL";
@@ -71,7 +70,7 @@ Set<Permission> parseTargetOperation(String inputOption, IResource resource)
         }
         catch (IllegalArgumentException e)
         {
-            throw appendCause(new InvalidRequestException("Invalid whitelist option: " + e.getMessage()), e);
+            throw Exceptions.appendCause(new InvalidRequestException("Invalid whitelist option: " + e.getMessage()), e);
         }
     }
 
@@ -93,7 +92,7 @@ IResource parseResource(String resourceName)
         }
         catch (IllegalArgumentException e)
         {
-            throw appendCause(new InvalidRequestException(String.format("Unable to parse whitelisted resource [%s]: %s", resourceName, e.getMessage())), e);
+            throw Exceptions.appendCause(new InvalidRequestException(String.format("Unable to parse whitelisted resource [%s]: %s", resourceName, e.getMessage())), e);
         }
     }
 }

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/config/AuditConfig.java

@@ -62,16 +62,25 @@ public ParameterizedClass getLoggerBackendParameters() throws ConfigurationExcep
         return yamlConfig.getLoggerBackendParameters();
     }
 
+    public boolean isPostLogging()
+    {
+        loadConfigIfNeeded();
+
+        return yamlConfig.isPostLogging();
+    }
+
+    public String getWrappedAuthorizer()
+    {
+        loadConfigIfNeeded();
+
+        return yamlConfig.getWrappedAuthorizer();
+    }
+
     private synchronized void loadConfigIfNeeded()
     {
         if (yamlConfig == null)
         {
             yamlConfig = yamlConfigurationLoader.loadConfig();
         }
     }
-
-    public boolean isPostLogging()
-    {
-        return yamlConfig.isPostLogging();
-    }
 }

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/config/AuditYamlConfig.java

@@ -30,6 +30,7 @@ public final class AuditYamlConfig
 {
     private static final List<String> DEFAULT_WHITELIST = Collections.emptyList();
     private static final ParameterizedClass DEFAULT_LOGGER_BACKEND = new ParameterizedClass(Slf4jAuditLogger.class.getCanonicalName(), Collections.emptyMap());
+    private static final String DEFAULT_WRAPPED_AUTHORIZER = "org.apache.cassandra.auth.CassandraAuthorizer";
 
     private boolean fromFile = true;
 
@@ -38,6 +39,7 @@ public final class AuditYamlConfig
     public List<String> whitelist;
     public ParameterizedClass logger_backend;
     public LoggerTiming log_timing_strategy;
+    public String wrapped_authorizer;
 
     static AuditYamlConfig createWithoutFile()
     {
@@ -92,4 +94,9 @@ boolean isPostLogging()
     {
         return log_timing_strategy == LoggerTiming.post_logging;
     }
+
+    String getWrappedAuthorizer()
+    {
+        return wrapped_authorizer != null ? wrapped_authorizer : DEFAULT_WRAPPED_AUTHORIZER;
+    }
 }

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/filter/role/RoleAuditFilter.java

@@ -25,7 +25,7 @@
 import com.google.common.util.concurrent.UncheckedExecutionException;
 
 import com.ericsson.bss.cassandra.ecaudit.auth.AuditWhitelistCache;
-import com.ericsson.bss.cassandra.ecaudit.auth.Exceptions;
+import com.ericsson.bss.cassandra.ecaudit.utils.Exceptions;
 import com.ericsson.bss.cassandra.ecaudit.auth.WhitelistDataAccess;
 import com.ericsson.bss.cassandra.ecaudit.entry.AuditEntry;
 import com.ericsson.bss.cassandra.ecaudit.filter.AuditFilter;

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/logger/ChronicleAuditLoggerConfig.java

@@ -21,12 +21,11 @@
 import java.util.Map;
 import java.util.Optional;
 
+import com.ericsson.bss.cassandra.ecaudit.utils.Exceptions;
 import net.openhft.chronicle.queue.RollCycle;
 import net.openhft.chronicle.queue.RollCycles;
 import org.apache.cassandra.exceptions.ConfigurationException;
 
-import static com.ericsson.bss.cassandra.ecaudit.auth.Exceptions.appendCause;
-
 class ChronicleAuditLoggerConfig
 {
     private static final String CONFIG_LOG_DIR = "log_dir";
@@ -83,7 +82,7 @@ private long resolveMaxLogSize(Map<String, String> parameters)
         }
         catch (NumberFormatException e)
         {
-            throw appendCause(new ConfigurationException("Invalid chronicle logger max log size: " + parameters.get(CONFIG_MAX_LOG_SIZE)), e);
+            throw Exceptions.appendCause(new ConfigurationException("Invalid chronicle logger max log size: " + parameters.get(CONFIG_MAX_LOG_SIZE)), e);
         }
 
         if (size <= 0)

ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/auth/Exceptions.java → ecaudit/src/main/java/com/ericsson/bss/cassandra/ecaudit/utils/Exceptions.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package com.ericsson.bss.cassandra.ecaudit.auth;
+package com.ericsson.bss.cassandra.ecaudit.utils;
 
 import org.apache.cassandra.exceptions.CassandraException;
 

ecaudit/src/test/java/com/ericsson/bss/cassandra/ecaudit/auth/TestAuditAuthorizer.java

@@ -26,6 +26,8 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import com.ericsson.bss.cassandra.ecaudit.config.AuditConfig;
+import org.apache.cassandra.auth.AllowAllAuthorizer;
 import org.apache.cassandra.auth.AuthenticatedUser;
 import org.apache.cassandra.auth.DataResource;
 import org.apache.cassandra.auth.IAuthorizer;
@@ -40,6 +42,7 @@
 import static org.assertj.core.api.Java6Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
@@ -212,6 +215,14 @@ public void testSetupDelegation()
         verify(mockAuthorizer).setup();
     }
 
+    @Test
+    public void testDynamicInstantiation()
+    {
+        AuditConfig auditConfig = givenConfigHasAllowAllAuthorizer();
+        IAuthorizer authorizer = AuditAuthorizer.newWrappedAuthorizer(auditConfig);
+        assertThat(authorizer).isInstanceOf(AllowAllAuthorizer.class);
+    }
+
     private Set<Permission> givenAuthorizedPermission(IAuthorizer authorizer, IResource resource)
     {
         Set<Permission> permissions = ImmutableSet.of(Permission.SELECT, Permission.MODIFY);
@@ -232,4 +243,11 @@ private Set<IResource> givenProtectedCassandraResources(IAuthorizer authorizer)
         doReturn(cassandraResources).when(authorizer).protectedResources();
         return cassandraResources;
     }
+
+    private AuditConfig givenConfigHasAllowAllAuthorizer()
+    {
+        AuditConfig auditConfig = mock(AuditConfig.class);
+        when(auditConfig.getWrappedAuthorizer()).thenReturn("org.apache.cassandra.auth.AllowAllAuthorizer");
+        return auditConfig;
+    }
 }

ecaudit/src/test/java/com/ericsson/bss/cassandra/ecaudit/config/TestAuditYamlConfigurationLoader.java

@@ -99,6 +99,26 @@ public void testLoadWhitelistAllPresent()
         assertThat(config.isPostLogging()).isTrue();
     }
 
+    @Test
+    public void testPostLoggingDefault()
+    {
+        Properties properties = getProperties("empty.yaml");
+
+        AuditConfig config = givenLoadedConfig(properties);
+
+        assertThat(config.isPostLogging()).isFalse();
+    }
+
+    @Test
+    public void testPostLoggingConfigured()
+    {
+        Properties properties = getProperties("mock_configuration.yaml");
+
+        AuditConfig config = givenLoadedConfig(properties);
+
+        assertThat(config.isPostLogging()).isTrue();
+    }
+
     @Test
     public void testLoggerTypeWhenMissingDefaultFile()
     {
@@ -120,6 +140,23 @@ public void testLoggerParametersWhenLoadingCustomConfiguration()
                                                                                 entry("max_log_size", "1000000"));
     }
 
+    @Test
+    public void testAuthorizerTypeWhenMissingDefaultFile()
+    {
+        AuditConfig config = givenLoadedDefaultConfig();
+
+        assertThat(config.getWrappedAuthorizer()).isEqualTo("org.apache.cassandra.auth.CassandraAuthorizer");
+    }
+
+    @Test
+    public void testCustomAuthorizerType()
+    {
+        Properties properties = getProperties("mock_configuration.yaml");
+        AuditConfig config = givenLoadedConfig(properties);
+
+        assertThat(config.getWrappedAuthorizer()).isEqualTo("org.apache.cassandra.auth.AllowAllAuthorizer");
+    }
+
     private AuditConfig givenLoadedConfig(Properties properties)
     {
         AuditYamlConfigurationLoader loader = AuditYamlConfigurationLoader.withProperties(properties);