Add kerberos auth support to kudu connector

core/trino-main/src/main/java/io/trino/server/security/KerberosAuthenticator.java

@@ -41,11 +41,10 @@
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
 
-import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.net.HttpHeaders.AUTHORIZATION;
+import static io.trino.plugin.base.util.SystemProperties.setJavaSecurityKrb5Conf;
 import static io.trino.server.security.UserMapping.createUserMapping;
 import static java.util.Objects.requireNonNull;
 import static javax.security.auth.login.AppConfigurationEntry.LoginModuleControlFlag.REQUIRED;
@@ -70,14 +69,7 @@ public KerberosAuthenticator(KerberosConfig config)
         requireNonNull(config, "config is null");
         this.userMapping = createUserMapping(config.getUserMappingPattern(), config.getUserMappingFile());
 
-        String newValue = config.getKerberosConfig().getAbsolutePath();
-        String currentValue = System.getProperty("java.security.krb5.conf");
-        checkState(
-                currentValue == null || Objects.equals(currentValue, newValue),
-                "Refusing to set system property 'java.security.krb5.conf' to '%s', it is already set to '%s'",
-                newValue,
-                currentValue);
-        System.setProperty("java.security.krb5.conf", newValue);
+        setJavaSecurityKrb5Conf(config.getKerberosConfig().getAbsolutePath());
 
         try {
             String hostname = Optional.ofNullable(config.getPrincipalHostname())

docs/src/main/sphinx/connector/kudu.rst

@@ -26,6 +26,9 @@ replacing the properties as appropriate:
 
    connector.name=kudu
 
+   ## Defaults to NONE
+   kudu.authentication.type = NONE
+
    ## List of Kudu master addresses, at least one is needed (comma separated)
    ## Supported formats: example.com, example.com:7051, 192.0.2.1, 192.0.2.1:7051,
    ##                    [2001:db8::1], [2001:db8::1]:7051, 2001:db8::1
@@ -54,6 +57,29 @@ replacing the properties as appropriate:
    ## Disable Kudu client's collection of statistics.
    #kudu.client.disable-statistics = false
 
+Kerberos support
+----------------
+
+In order to connect to a kudu cluster that uses ``kerberos``
+authentication, you need to configure the following kudu properties:
+
+.. code-block:: properties
+
+   kudu.authentication.type = KERBEROS
+
+   ## The kerberos client principal name
+   kudu.authentication.client.principal = clientprincipalname
+
+   ## The path to the kerberos keytab file
+   ## The configured client principal must exist in this keytab file
+   kudu.authentication.client.keytab = /path/to/keytab/file.keytab
+
+   ## The path to the krb5.conf kerberos config file
+   kudu.authentication.config = /path/to/kerberos/krb5.conf
+
+   ## Optional and defaults to "kudu"
+   ## If kudu is running with a custom SPN this needs to be configured
+   kudu.authentication.server.principal.primary = kudu
 
 Querying data
 -------------
@@ -588,4 +614,3 @@ Limitations
 -----------
 
 -  Only lower case table and column names in Kudu are supported.
--  Using a secured Kudu cluster has not been tested.

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/authentication/CachingKerberosAuthentication.java

@@ -37,11 +37,26 @@ public CachingKerberosAuthentication(KerberosAuthentication kerberosAuthenticati
 
     public synchronized Subject getSubject()
     {
-        if (subject == null || nextRefreshTime < System.currentTimeMillis()) {
+        if (subject == null || ticketNeedsRefresh()) {
             subject = requireNonNull(kerberosAuthentication.getSubject(), "kerberosAuthentication.getSubject() is null");
             KerberosTicket tgtTicket = getTicketGrantingTicket(subject);
             nextRefreshTime = KerberosTicketUtils.getRefreshTime(tgtTicket);
         }
         return subject;
     }
+
+    public synchronized void reauthenticateIfSoonWillBeExpired()
+    {
+        requireNonNull(subject, "subject is null, getSubject() must be called before reauthenticate()");
+        if (ticketNeedsRefresh()) {
+            kerberosAuthentication.attemptLogin(subject);
+            KerberosTicket tgtTicket = getTicketGrantingTicket(subject);
+            nextRefreshTime = KerberosTicketUtils.getRefreshTime(tgtTicket);
+        }
+    }
+
+    private boolean ticketNeedsRefresh()
+    {
+        return nextRefreshTime < System.currentTimeMillis();
+    }
 }

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/authentication/KerberosAuthentication.java

@@ -73,6 +73,17 @@ public Subject getSubject()
         }
     }
 
+    public void attemptLogin(Subject subject)
+    {
+        try {
+            LoginContext loginContext = new LoginContext("", subject, null, configuration);
+            loginContext.login();
+        }
+        catch (LoginException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     private static KerberosPrincipal createKerberosPrincipal(String principal)
     {
         try {

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/util/SystemProperties.java

@@ -0,0 +1,47 @@
+/*
+ * 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 io.trino.plugin.base.util;
+
+import java.util.Objects;
+
+import static com.google.common.base.Preconditions.checkState;
+
+public final class SystemProperties
+{
+    public static final String JAVA_SECURITY_KRB5_CONF = "java.security.krb5.conf";
+
+    private SystemProperties() {}
+
+    public static synchronized void setJavaSecurityKrb5Conf(String value)
+    {
+        set(JAVA_SECURITY_KRB5_CONF, value);
+    }
+
+    public static synchronized void set(String key, String newValue)
+    {
+        String currentValue = System.getProperty(key);
+        checkSameValues(key, newValue, currentValue);
+        System.setProperty(key, newValue);
+    }
+
+    private static void checkSameValues(String key, String newValue, String currentValue)
+    {
+        checkState(
+                currentValue == null || Objects.equals(currentValue, newValue),
+                "Refusing to set system property '%s' to '%s', it is already set to '%s'",
+                key,
+                newValue,
+                currentValue);
+    }
+}

plugin/trino-kudu/pom.xml

@@ -134,6 +134,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-spi</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-testing</artifactId>

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/ForwardingKuduClient.java

@@ -0,0 +1,116 @@
+/*
+ * 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 io.trino.plugin.kudu;
+
+import org.apache.kudu.Schema;
+import org.apache.kudu.client.AlterTableOptions;
+import org.apache.kudu.client.AlterTableResponse;
+import org.apache.kudu.client.CreateTableOptions;
+import org.apache.kudu.client.DeleteTableResponse;
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduException;
+import org.apache.kudu.client.KuduScanToken;
+import org.apache.kudu.client.KuduScanner;
+import org.apache.kudu.client.KuduSession;
+import org.apache.kudu.client.KuduTable;
+import org.apache.kudu.client.ListTablesResponse;
+
+import java.io.IOException;
+
+public abstract class ForwardingKuduClient
+        implements KuduClientWrapper
+{
+    protected abstract KuduClient delegate();
+
+    @Override
+    public KuduTable createTable(String name, Schema schema, CreateTableOptions builder)
+            throws KuduException
+    {
+        return delegate().createTable(name, schema, builder);
+    }
+
+    @Override
+    public DeleteTableResponse deleteTable(String name)
+            throws KuduException
+    {
+        return delegate().deleteTable(name);
+    }
+
+    @Override
+    public AlterTableResponse alterTable(String name, AlterTableOptions ato)
+            throws KuduException
+    {
+        return delegate().alterTable(name, ato);
+    }
+
+    @Override
+    public ListTablesResponse getTablesList()
+            throws KuduException
+    {
+        return delegate().getTablesList();
+    }
+
+    @Override
+    public ListTablesResponse getTablesList(String nameFilter)
+            throws KuduException
+    {
+        return delegate().getTablesList(nameFilter);
+    }
+
+    @Override
+    public boolean tableExists(String name)
+            throws KuduException
+    {
+        return delegate().tableExists(name);
+    }
+
+    @Override
+    public KuduTable openTable(final String name)
+            throws KuduException
+    {
+        return delegate().openTable(name);
+    }
+
+    @Override
+    public KuduScanner.KuduScannerBuilder newScannerBuilder(KuduTable table)
+    {
+        return delegate().newScannerBuilder(table);
+    }
+
+    @Override
+    public KuduScanToken.KuduScanTokenBuilder newScanTokenBuilder(KuduTable table)
+    {
+        return delegate().newScanTokenBuilder(table);
+    }
+
+    @Override
+    public KuduSession newSession()
+    {
+        return delegate().newSession();
+    }
+
+    @Override
+    public KuduScanner deserializeIntoScanner(byte[] serializedScanToken)
+            throws IOException
+    {
+        return KuduScanToken.deserializeIntoScanner(serializedScanToken, delegate());
+    }
+
+    @Override
+    public void close()
+            throws KuduException
+    {
+        delegate().close();
+    }
+}

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KerberizedKuduClient.java

@@ -0,0 +1,45 @@
+/*
+ * 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 io.trino.plugin.kudu;
+
+import io.trino.plugin.base.authentication.CachingKerberosAuthentication;
+import org.apache.kudu.client.KuduClient;
+
+import javax.security.auth.Subject;
+
+import java.security.PrivilegedAction;
+
+import static java.util.Objects.requireNonNull;
+import static org.apache.kudu.client.KuduClient.KuduClientBuilder;
+
+public class KerberizedKuduClient
+        extends ForwardingKuduClient
+{
+    private final KuduClient kuduClient;
+    private final CachingKerberosAuthentication cachingKerberosAuthentication;
+
+    KerberizedKuduClient(KuduClientBuilder kuduClientBuilder, CachingKerberosAuthentication cachingKerberosAuthentication)
+    {
+        requireNonNull(kuduClientBuilder, "kuduClientBuilder is null");
+        this.cachingKerberosAuthentication = requireNonNull(cachingKerberosAuthentication, "cachingKerberosAuthentication is null");
+        kuduClient = Subject.doAs(cachingKerberosAuthentication.getSubject(), (PrivilegedAction<KuduClient>) (kuduClientBuilder::build));
+    }
+
+    @Override
+    protected KuduClient delegate()
+    {
+        cachingKerberosAuthentication.reauthenticateIfSoonWillBeExpired();
+        return kuduClient;
+    }
+}

plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduAuthenticationConfig.java

@@ -0,0 +1,41 @@
+/*
+ * 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 io.trino.plugin.kudu;
+
+import io.airlift.configuration.Config;
+import io.airlift.configuration.ConfigDescription;
+
+public class KuduAuthenticationConfig
+{
+    private KuduAuthenticationType authenticationType = KuduAuthenticationType.NONE;
+
+    public enum KuduAuthenticationType
+    {
+        NONE,
+        KERBEROS;
+    }
+
+    public KuduAuthenticationType getAuthenticationType()
+    {
+        return authenticationType;
+    }
+
+    @Config("kudu.authentication.type")
+    @ConfigDescription("Kudu authentication type")
+    public KuduAuthenticationConfig setAuthenticationType(KuduAuthenticationType authenticationType)
+    {
+        this.authenticationType = authenticationType;
+        return this;
+    }
+}