Add support for Kerberos AuthProvider

Author: maximevw

Diff for: CHANGELOG.md

@@ -9,6 +9,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Add a parameter `fetchsize` to specify a default fetch size for all the queries returning result sets. This value is 
   the number of rows the server will return in each network frame (see 
   [paging documentation](https://docs.datastax.com/en/developer/java-driver/latest/manual/core/paging/)).
+- Add support for Kerberos authentication provider.
 ### Changed
 - Modify the types of some columns in the result sets of the following methods of `CassandraDatabaseMetadata` to respect
   the JDBC API specifications:

Diff for: pom.xml

@@ -114,6 +114,7 @@
         <!-- Versions for dependencies -->
         <checkstyle.version>9.3</checkstyle.version>
         <caffeine.version>2.9.3</caffeine.version>
+        <cassandra-driver-krb5.version>3.0.0</cassandra-driver-krb5.version>
         <commons-collections.version>4.4</commons-collections.version>
         <commons-io.version>2.15.1</commons-io.version>
         <commons-lang3.version>3.14.0</commons-lang3.version>
@@ -201,6 +202,13 @@
             <version>${semver4j.version}</version>
         </dependency>
 
+        <!-- Kerberos Auth provider support -->
+        <dependency>
+            <groupId>com.instaclustr</groupId>
+            <artifactId>cassandra-driver-kerberos</artifactId>
+            <version>${cassandra-driver-krb5.version}</version>
+        </dependency>
+
         <!-- Unit tests libraries -->
         <dependency>
             <groupId>org.mockito</groupId>

Diff for: src/main/java/com/ing/data/cassandra/jdbc/SessionHolder.java

@@ -41,6 +41,8 @@
 import com.ing.data.cassandra.jdbc.codec.TinyintToIntCodec;
 import com.ing.data.cassandra.jdbc.codec.VarintToIntCodec;
 import com.ing.data.cassandra.jdbc.utils.ContactPoint;
+import com.instaclustr.cassandra.driver.auth.KerberosAuthProviderBase;
+import com.instaclustr.cassandra.driver.auth.ProgrammaticKerberosAuthProvider;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.math.NumberUtils;
 import org.slf4j.Logger;
@@ -65,6 +67,7 @@
 import static com.ing.data.cassandra.jdbc.utils.DriverUtil.JSSE_TRUSTSTORE_PASSWORD_PROPERTY;
 import static com.ing.data.cassandra.jdbc.utils.DriverUtil.JSSE_TRUSTSTORE_PROPERTY;
 import static com.ing.data.cassandra.jdbc.utils.ErrorConstants.SSL_CONFIG_FAILED;
+import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_USE_KERBEROS;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CLOUD_SECURE_CONNECT_BUNDLE;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONFIG_FILE;
 import static com.ing.data.cassandra.jdbc.utils.JdbcUrlUtil.TAG_CONNECT_TIMEOUT;
@@ -203,6 +206,7 @@ private Session createSession(final Properties properties) throws SQLException {
                 this.properties.remove(TAG_CONNECT_TIMEOUT);
                 this.properties.remove(TAG_KEEP_ALIVE);
                 this.properties.remove(TAG_TCP_NO_DELAY);
+                this.properties.remove(TAG_USE_KERBEROS);
                 LOG.info("The configuration file {} will be used and will override the parameters defined into the "
                     + "JDBC URL except contact points and keyspace.", configurationFilePath);
             } else {
@@ -234,6 +238,8 @@ private Session createSession(final Properties properties) throws SQLException {
         if (NumberUtils.isParsable(requestTimeoutRawValue)) {
             requestTimeout = Integer.parseInt(requestTimeoutRawValue);
         }
+        final boolean useKerberosAuthProvider = Boolean.TRUE.toString().equals(properties.getProperty(TAG_USE_KERBEROS,
+            StringUtils.EMPTY));
 
         // Instantiate the session builder and set the contact points.
         final CqlSessionBuilder builder = CqlSession.builder();
@@ -328,6 +334,13 @@ private Session createSession(final Properties properties) throws SQLException {
             }
         }
 
+        // Set Kerberos Auth provider if required.
+        if (useKerberosAuthProvider) {
+            builder.withAuthProvider(new ProgrammaticKerberosAuthProvider(
+                KerberosAuthProviderBase.KerberosAuthOptions.builder().build()
+            ));
+        }
+
         // Declare and register codecs.
         final List<TypeCodec<?>> codecs = new ArrayList<>();
         codecs.add(new TimestampToLongCodec());

Diff for: src/main/java/com/ing/data/cassandra/jdbc/utils/JdbcUrlUtil.java

@@ -223,6 +223,17 @@ public final class JdbcUrlUtil {
      */
     public static final String TAG_PASSWORD = "password";
 
+    /**
+     * JDBC URL parameter key for Kerberos auth provider enabling.
+     */
+    public static final String KEY_USE_KERBEROS = "usekrb5";
+
+    /**
+     * Property name used to retrieve the Kerberos auth provider enabling when the connection to Cassandra is
+     * established. This property is mapped from the JDBC URL parameter {@code usekrb5}.
+     */
+    public static final String TAG_USE_KERBEROS = "useKerberos";
+
     /**
      * JDBC URL parameter key for the request timeout.
      */
@@ -420,6 +431,9 @@ public static Properties parseURL(final String url) throws SQLException {
                 if (params.containsKey(KEY_PASSWORD)) {
                     props.setProperty(TAG_PASSWORD, params.get(KEY_PASSWORD));
                 }
+                if (params.containsKey(KEY_USE_KERBEROS)) {
+                    props.setProperty(TAG_USE_KERBEROS, params.get(KEY_USE_KERBEROS));
+                }
                 if (params.containsKey(KEY_REQUEST_TIMEOUT)) {
                     props.setProperty(TAG_REQUEST_TIMEOUT, params.get(KEY_REQUEST_TIMEOUT));
                 }

Diff for: src/test/java/com/ing/data/cassandra/jdbc/KerberosAuthProviderTest.java

@@ -0,0 +1,53 @@
+/*
+ *   Licensed 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 com.ing.data.cassandra.jdbc;
+
+import com.datastax.oss.driver.api.core.auth.AuthProvider;
+import com.instaclustr.cassandra.driver.auth.KerberosAuthProviderBase;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+/**
+ * Test connection using Kerberos Auth provider.
+ */
+class KerberosAuthProviderTest extends UsingCassandraContainerTest {
+
+    private static final String KEYSPACE = "system";
+
+    @BeforeAll
+    static void finalizeSetUpTests() throws Exception {
+        initConnection(KEYSPACE, "localdatacenter=datacenter1", "usekrb5=true");
+    }
+
+    @Test
+    void givenValidConnectionString_whenGetConnection_createConnectionWithKrb5Provider() throws Exception {
+        assertNotNull(sqlConnection);
+        assertNotNull(sqlConnection.getSession());
+        assertNotNull(sqlConnection.getSession().getContext());
+        final Optional<AuthProvider> configuredAuthProvider = sqlConnection.getSession().getContext().getAuthProvider();
+        assertTrue(configuredAuthProvider.isPresent());
+        assertTrue(KerberosAuthProviderBase.class.isAssignableFrom(configuredAuthProvider.get().getClass()));
+        assertTrue(sqlConnection.getSession().getKeyspace().isPresent());
+        assertEquals(KEYSPACE, sqlConnection.getSession().getKeyspace().get().asCql(true));
+        sqlConnection.close();
+    }
+
+}