[core] Introduce SecurityContext

docs/content/filesystems/hdfs.md

@@ -77,14 +77,50 @@ For Alluxio support add the following entry into the core-site.xml file:
 </property>
 ```
 
+## Kerberos
+
+{{< tabs "Kerberos" >}}
+
+{{< tab "Flink" >}}
+
+It is recommented to use [Flink Kerberos Keytab](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/security/security-kerberos/).
+
+{{< /tab >}}
+
+{{< tab "Spark" >}}
+
+It is recommented to use [Spark Kerberos Keytab](https://spark.apache.org/docs/latest/security.html#using-a-keytab).
+
+{{< /tab >}}
+
+{{< tab "Hive" >}}
+
+An intuitive approach is to configure Hive's kerberos authentication.
+
+{{< /tab >}}
+
+{{< tab "Trino/JavaAPI" >}}
+
+Configure the following three options in your catalog configuration:
+
+- security.kerberos.login.keytabs: Absolute path to a Kerberos keytab file that contains the user credentials.
+  Please make sure it is copied to each machine.
+- security.kerberos.login.principal: Kerberos principal name associated with the keytab.
+- security.kerberos.login.use-ticket-cache: True or false, indicates whether to read from your Kerberos ticket cache.
+
+For JavaAPI:
+```
+SecurityContext.install(catalogOptions);
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
 ## HDFS HA
 
 Ensure that `hdfs-site.xml` and `core-site.xml` contain the necessary [HA configuration](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HDFSHighAvailabilityWithNFS.html).
 
 ## HDFS ViewFS
 
 Ensure that `hdfs-site.xml` and `core-site.xml` contain the necessary [ViewFs configuration](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/ViewFs.html).
-
-## Kerberos
-
-Ensure that `hdfs-site.xml` and `core-site.xml` contain the necessary [Kerberos configuration](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/SecureMode.html).

paimon-common/src/main/java/org/apache/paimon/security/HadoopModule.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.security;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.apache.paimon.security.KerberosLoginProvider.hasUserKerberosAuthMethod;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Responsible for installing a Hadoop login user. */
+public class HadoopModule {
+
+    private static final Logger LOG = LoggerFactory.getLogger(HadoopModule.class);
+
+    private final SecurityConfiguration securityConfig;
+
+    private final Configuration hadoopConfiguration;
+
+    public HadoopModule(
+            SecurityConfiguration securityConfiguration, Configuration hadoopConfiguration) {
+        this.securityConfig = checkNotNull(securityConfiguration);
+        this.hadoopConfiguration = checkNotNull(hadoopConfiguration);
+    }
+
+    public void install() throws IOException {
+
+        UserGroupInformation.setConfiguration(hadoopConfiguration);
+
+        UserGroupInformation loginUser;
+
+        KerberosLoginProvider kerberosLoginProvider = new KerberosLoginProvider(securityConfig);
+        if (kerberosLoginProvider.isLoginPossible()) {
+            kerberosLoginProvider.doLogin();
+            loginUser = UserGroupInformation.getLoginUser();
+
+            if (loginUser.isFromKeytab()) {
+                String fileLocation =
+                        System.getenv(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION);
+                if (fileLocation != null) {
+                    Credentials credentials =
+                            Credentials.readTokenStorageFile(
+                                    new File(fileLocation), hadoopConfiguration);
+                    loginUser.addCredentials(credentials);
+                }
+            }
+        } else {
+            loginUser = UserGroupInformation.getLoginUser();
+        }
+
+        LOG.info("Hadoop user set to {}", loginUser);
+        boolean isKerberosSecurityEnabled = hasUserKerberosAuthMethod(loginUser);
+        LOG.info("Kerberos security is {}.", isKerberosSecurityEnabled ? "enabled" : "disabled");
+        if (isKerberosSecurityEnabled) {
+            LOG.info(
+                    "Kerberos credentials are {}.",
+                    loginUser.hasKerberosCredentials() ? "valid" : "invalid");
+        }
+    }
+}

paimon-common/src/main/java/org/apache/paimon/security/HadoopSecurityContext.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.security;
+
+import org.apache.hadoop.security.UserGroupInformation;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.concurrent.Callable;
+
+/**
+ * Hadoop security context which runs a Callable with the previously initialized UGI and appropriate
+ * security credentials.
+ */
+public class HadoopSecurityContext {
+
+    private final UserGroupInformation ugi;
+
+    public HadoopSecurityContext() throws IOException {
+        this.ugi = UserGroupInformation.getLoginUser();
+    }
+
+    public <T> T runSecured(final Callable<T> securedCallable) throws Exception {
+        return ugi.doAs((PrivilegedExceptionAction<T>) securedCallable::call);
+    }
+}

paimon-common/src/main/java/org/apache/paimon/security/KerberosLoginProvider.java

@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.security;
+
+import org.apache.paimon.options.Options;
+
+import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Provides Kerberos login functionality. */
+public class KerberosLoginProvider {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KerberosLoginProvider.class);
+
+    private final String principal;
+
+    private final String keytab;
+
+    private final boolean useTicketCache;
+
+    public KerberosLoginProvider(Options options) {
+        checkNotNull(options, "options must not be null");
+        SecurityConfiguration securityConfiguration = new SecurityConfiguration(options);
+        this.principal = securityConfiguration.getPrincipal();
+        this.keytab = securityConfiguration.getKeytab();
+        this.useTicketCache = securityConfiguration.useTicketCache();
+    }
+
+    public KerberosLoginProvider(SecurityConfiguration config) {
+        checkNotNull(config, "SecurityConfiguration must not be null");
+        this.principal = config.getPrincipal();
+        this.keytab = config.getKeytab();
+        this.useTicketCache = config.useTicketCache();
+    }
+
+    public boolean isLoginPossible() throws IOException {
+        if (UserGroupInformation.isSecurityEnabled()) {
+            LOG.debug("Security is enabled");
+        } else {
+            LOG.debug("Security is NOT enabled");
+            return false;
+        }
+
+        UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
+
+        if (principal != null) {
+            LOG.debug("Login from keytab is possible");
+            return true;
+        } else if (!isProxyUser(currentUser)) {
+            if (useTicketCache && currentUser.hasKerberosCredentials()) {
+                LOG.debug("Login from ticket cache is possible");
+                return true;
+            }
+        } else {
+            throwProxyUserNotSupported();
+        }
+
+        LOG.debug("Login is NOT possible");
+
+        return false;
+    }
+
+    /**
+     * Does kerberos login and sets current user. Must be called when isLoginPossible returns true.
+     */
+    public void doLogin() throws IOException {
+        if (principal != null) {
+            LOG.info(
+                    "Attempting to login to KDC using principal: {} keytab: {}", principal, keytab);
+            UserGroupInformation.loginUserFromKeytab(principal, keytab);
+            LOG.info("Successfully logged into KDC");
+        } else if (!isProxyUser(UserGroupInformation.getCurrentUser())) {
+            LOG.info("Attempting to load user's ticket cache");
+            UserGroupInformation.loginUserFromSubject(null);
+            LOG.info("Loaded user's ticket cache successfully");
+        } else {
+            throwProxyUserNotSupported();
+        }
+    }
+
+    private void throwProxyUserNotSupported() {
+        throw new UnsupportedOperationException("Proxy user is not supported");
+    }
+
+    public static boolean isProxyUser(UserGroupInformation ugi) {
+        return ugi.getAuthenticationMethod() == UserGroupInformation.AuthenticationMethod.PROXY;
+    }
+
+    public static boolean hasUserKerberosAuthMethod(UserGroupInformation ugi) {
+        return UserGroupInformation.isSecurityEnabled()
+                && ugi.getAuthenticationMethod()
+                        == UserGroupInformation.AuthenticationMethod.KERBEROS;
+    }
+}

paimon-common/src/main/java/org/apache/paimon/security/SecurityConfiguration.java

@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.paimon.security;
+
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.StringUtils;
+
+import java.io.File;
+
+import static org.apache.paimon.options.ConfigOptions.key;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** The security configuration. */
+public class SecurityConfiguration {
+
+    public static final ConfigOption<String> KERBEROS_LOGIN_KEYTAB =
+            key("security.kerberos.login.keytab")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDeprecatedKeys("security.keytab")
+                    .withDescription(
+                            "Absolute path to a Kerberos keytab file that contains the user credentials.");
+
+    public static final ConfigOption<String> KERBEROS_LOGIN_PRINCIPAL =
+            key("security.kerberos.login.principal")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDeprecatedKeys("security.principal")
+                    .withDescription("Kerberos principal name associated with the keytab.");
+
+    public static final ConfigOption<Boolean> KERBEROS_LOGIN_USETICKETCACHE =
+            key("security.kerberos.login.use-ticket-cache")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription("Indicates whether to read from your Kerberos ticket cache.");
+
+    private final Options options;
+
+    private final boolean useTicketCache;
+
+    private final String keytab;
+
+    private final String principal;
+
+    public SecurityConfiguration(Options options) {
+        this.options = checkNotNull(options);
+        this.keytab = options.get(KERBEROS_LOGIN_KEYTAB);
+        this.principal = options.get(KERBEROS_LOGIN_PRINCIPAL);
+        this.useTicketCache = options.get(KERBEROS_LOGIN_USETICKETCACHE);
+    }
+
+    public String getKeytab() {
+        return keytab;
+    }
+
+    public String getPrincipal() {
+        return principal;
+    }
+
+    public boolean useTicketCache() {
+        return useTicketCache;
+    }
+
+    public Options getOptions() {
+        return options;
+    }
+
+    public boolean isLegal() {
+        if (StringUtils.isBlank(keytab) != StringUtils.isBlank(principal)) {
+            return false;
+        }
+
+        if (!StringUtils.isBlank(keytab)) {
+            File keytabFile = new File(keytab);
+            return keytabFile.exists() && keytabFile.isFile() && keytabFile.canRead();
+        }
+
+        return true;
+    }
+}