[
https://issues.apache.org/jira/browse/FLINK-3929?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15503261#comment-15503261
]
ASF GitHub Bot commented on FLINK-3929:
---------------------------------------
Github user mxm commented on a diff in the pull request:
https://github.com/apache/flink/pull/2275#discussion_r79355908
--- Diff:
flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/SecureTestEnvironment.java
---
@@ -0,0 +1,249 @@
+/*
+ * 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.flink.test.util;
+
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.security.SecurityContext;
+import org.apache.hadoop.minikdc.MiniKdc;
+import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.BufferedWriter;
+import java.io.PrintWriter;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Helper {@link SecureTestEnvironment} to handle MiniKDC lifecycle.
+ * This class can be used to start/stop MiniKDC and create secure
configurations for MiniDFSCluster
+ * and MiniYarn
+ */
+
+public class SecureTestEnvironment {
+
+ protected static final Logger LOG =
LoggerFactory.getLogger(SecureTestEnvironment.class);
+
+ private static MiniKdc kdc;
+
+ private static String testKeytab = null;
+
+ private static String testPrincipal = null;
+
+ private static String testZkServerPrincipal = null;
+
+ private static String testZkClientPrincipal = null;
+
+ private static String testKafkaServerPrincipal = null;
+
+ private static String hadoopServicePrincipal = null;
+
+ private static File baseDirForSecureRun = null;
+
+ public static void prepare(TemporaryFolder tempFolder) {
+
+ try {
+ baseDirForSecureRun = tempFolder.newFolder();
+ LOG.info("Base Directory for Secure Environment: {}",
baseDirForSecureRun);
+
+ String hostName = "localhost";
+ Properties kdcConf = MiniKdc.createConf();
+ if(LOG.isDebugEnabled()) {
+ kdcConf.setProperty(MiniKdc.DEBUG, "true");
+ }
+ kdcConf.setProperty(MiniKdc.KDC_BIND_ADDRESS, hostName);
+ kdc = new MiniKdc(kdcConf, baseDirForSecureRun);
+ kdc.start();
+ LOG.info("Started Mini KDC");
+
+ File keytabFile = new File(baseDirForSecureRun,
"test-users.keytab");
+ testKeytab = keytabFile.getAbsolutePath();
+ testZkServerPrincipal = "zookeeper/127.0.0.1";
+ testZkClientPrincipal = "zk-client/127.0.0.1";
+ testKafkaServerPrincipal = "kafka/" + hostName;
+ hadoopServicePrincipal = "hadoop/" + hostName;
+ testPrincipal = "client/" + hostName;
+
+ kdc.createPrincipal(keytabFile, testPrincipal,
testZkServerPrincipal,
+ hadoopServicePrincipal,
+ testZkClientPrincipal,
+ testKafkaServerPrincipal);
+
+ testPrincipal = testPrincipal + "@" + kdc.getRealm();
+ testZkServerPrincipal = testZkServerPrincipal + "@" +
kdc.getRealm();
+ testZkClientPrincipal = testZkClientPrincipal + "@" +
kdc.getRealm();
+ testKafkaServerPrincipal = testKafkaServerPrincipal +
"@" + kdc.getRealm();
+ hadoopServicePrincipal = hadoopServicePrincipal + "@" +
kdc.getRealm();
+
+
LOG.info("-------------------------------------------------------------------");
+ LOG.info("Test Principal: {}", testPrincipal);
+ LOG.info("Test ZK Server Principal: {}",
testZkServerPrincipal);
+ LOG.info("Test ZK Client Principal: {}",
testZkClientPrincipal);
+ LOG.info("Test Kafka Server Principal: {}",
testKafkaServerPrincipal);
+ LOG.info("Test Hadoop Service Principal: {}",
hadoopServicePrincipal);
+ LOG.info("Test Keytab: {}", testKeytab);
+
LOG.info("-------------------------------------------------------------------");
+
+ //Security Context is established to allow non hadoop
applications that requires JAAS
+ //based SASL/Kerberos authentication to work. However,
for Hadoop specific applications
+ //the context can be reinitialized with Hadoop
configuration by calling
+ //ctx.setHadoopConfiguration() for the UGI
implementation to work properly.
+ //See Yarn test case module for reference
+ createJaasConfig(baseDirForSecureRun);
+ SecurityContext.SecurityConfiguration ctx = new
SecurityContext.SecurityConfiguration();
+ Configuration flinkConfig = new Configuration();
+
flinkConfig.setString(ConfigConstants.SECURITY_KEYTAB_KEY, testKeytab);
+
flinkConfig.setString(ConfigConstants.SECURITY_PRINCIPAL_KEY, testPrincipal);
+
flinkConfig.setBoolean(ConfigConstants.ZOOKEEPER_SASL_DISABLE, false);
+
flinkConfig.setString(ConfigConstants.FLINK_BASE_DIR_PATH_KEY,
baseDirForSecureRun.getAbsolutePath());
+ ctx.setFlinkConfiguration(flinkConfig);
+ TestingSecurityContext.install(ctx,
getClientSecurityConfigurationMap());
+
+ populateSystemEnvVariables();
+
+ } catch(Exception e) {
+ LOG.error("Exception occured while preparing secure
environment. Reason: {}", e);
+ throw new RuntimeException(e);
+ }
+
+ }
+
+ public static void cleanup() {
+
+ LOG.info("Cleaning up Secure Environment");
+
+ if( kdc != null) {
+ kdc.stop();
+ LOG.info("Stopped KDC server");
+ }
+
+ resetSystemEnvVariables();
+
+ testKeytab = null;
+ testPrincipal = null;
+ testZkServerPrincipal = null;
+ hadoopServicePrincipal = null;
+ baseDirForSecureRun = null;
+
+ }
+
+ private static void populateSystemEnvVariables() {
+
+ if(LOG.isDebugEnabled()) {
+ System.setProperty("FLINK_JAAS_DEBUG", "true");
+ System.setProperty("sun.security.krb5.debug", "true");
+ }
+
+ System.setProperty("java.security.krb5.conf",
kdc.getKrb5conf().getAbsolutePath());
+
+ System.setProperty("zookeeper.authProvider.1",
"org.apache.zookeeper.server.auth.SASLAuthenticationProvider");
+
System.setProperty("zookeeper.kerberos.removeHostFromPrincipal", "true");
+
System.setProperty("zookeeper.kerberos.removeRealmFromPrincipal", "true");
+ }
+
+ private static void resetSystemEnvVariables() {
+ System.clearProperty("java.security.krb5.conf");
+ System.clearProperty("FLINK_JAAS_DEBUG");
+ System.clearProperty("sun.security.krb5.debug");
+
+ System.clearProperty("zookeeper.authProvider.1");
+
System.clearProperty("zookeeper.kerberos.removeHostFromPrincipal");
+
System.clearProperty("zookeeper.kerberos.removeRealmFromPrincipal");
+ }
+
+ public static org.apache.flink.configuration.Configuration
populateFlinkSecureConfigurations(
+ @Nullable org.apache.flink.configuration.Configuration
flinkConf) {
+
+ org.apache.flink.configuration.Configuration conf;
+
+ if(flinkConf== null) {
+ conf = new
org.apache.flink.configuration.Configuration();
+ } else {
+ conf = flinkConf;
+ }
+
+ conf.setString(ConfigConstants.SECURITY_KEYTAB_KEY ,
testKeytab);
+ conf.setString(ConfigConstants.SECURITY_PRINCIPAL_KEY ,
testPrincipal);
+
+ return conf;
+ }
+
+ public static Map<String,
TestingSecurityContext.ClientSecurityConfiguration>
getClientSecurityConfigurationMap() {
+
+ Map<String, TestingSecurityContext.ClientSecurityConfiguration>
clientSecurityConfigurationMap = new HashMap<>();
+
+ if(testZkServerPrincipal != null ) {
+ TestingSecurityContext.ClientSecurityConfiguration
zkServer =
+ new
TestingSecurityContext.ClientSecurityConfiguration(testZkServerPrincipal,
testKeytab,
+ "Server", "zk-server");
+ clientSecurityConfigurationMap.put("Server",zkServer);
+ }
+
+ if(testZkClientPrincipal != null ) {
+ TestingSecurityContext.ClientSecurityConfiguration
zkClient =
+ new
TestingSecurityContext.ClientSecurityConfiguration(testZkClientPrincipal,
testKeytab,
+ "Client", "zk-client");
+ clientSecurityConfigurationMap.put("Client",zkClient);
+ }
+
+ if(testKafkaServerPrincipal != null ) {
+ TestingSecurityContext.ClientSecurityConfiguration
kafkaServer =
+ new
TestingSecurityContext.ClientSecurityConfiguration(testKafkaServerPrincipal,
testKeytab,
+ "KafkaServer",
"kafka-server");
+
clientSecurityConfigurationMap.put("KafkaServer",kafkaServer);
+ }
+
+ return clientSecurityConfigurationMap;
+ }
+
+ public static String getTestKeytab() {
+ return testKeytab;
+ }
+
+ public static String getHadoopServicePrincipal() {
+ return hadoopServicePrincipal;
+ }
+
+ /*
+ * Helper method to create a temporary JAAS configuration file to ger
around the Kafka and ZK SASL
+ * implementation lookup java.security.auth.login.config
+ */
+ private static void createJaasConfig(File baseDirForSecureRun) {
+
+ try(FileWriter fw = new FileWriter(new
File(baseDirForSecureRun,SecurityContext.JAAS_CONF_FILENAME), true);
+ BufferedWriter bw = new BufferedWriter(fw);
+ PrintWriter out = new PrintWriter(bw))
+ {
+ out.println("sample {");
+ out.println("useKeyTab=false");
+ out.println("useTicketCache=true;");
+ out.println("};");
+ } catch (IOException e) {
+ LOG.error("Exception occured while trying to create
JAAS config. Reason: {}", e.getMessage());
--- End diff --
Should be changed to `LOG.error("Exception occured while trying to create
JAAS config.", e);`
> Support for Kerberos Authentication with Keytab Credential
> ----------------------------------------------------------
>
> Key: FLINK-3929
> URL: https://issues.apache.org/jira/browse/FLINK-3929
> Project: Flink
> Issue Type: New Feature
> Reporter: Eron Wright
> Assignee: Vijay Srinivasaraghavan
> Labels: kerberos, security
> Original Estimate: 672h
> Remaining Estimate: 672h
>
> _This issue is part of a series of improvements detailed in the [Secure Data
> Access|https://docs.google.com/document/d/1-GQB6uVOyoaXGwtqwqLV8BHDxWiMO2WnVzBoJ8oPaAs/edit?usp=sharing]
> design doc._
> Add support for a keytab credential to be associated with the Flink cluster,
> to facilitate:
> - Kerberos-authenticated data access for connectors
> - Kerberos-authenticated ZooKeeper access
> Support both the standalone and YARN deployment modes.
>
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
