You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by bb...@apache.org on 2020/03/02 16:29:21 UTC
[nifi] branch master updated: NIFI-7025: Initial commit adding
Kerberos Password feature for Hive components Kerberos Password property
should not support EL,
this includes a change to KerberosProperties which is also used by the HDFS
processors (AbstractHadoopProcessor) Added wiring in a KerberosContext to a
TestRunner's MockProcessorInitializationContext Removed synchronization
blocks around KerberosUser.checkTGTAndRelogin,
since that method is already synchronized Updated AbstractHadoopProcessor to
have a boolean ac [...]
This is an automated email from the ASF dual-hosted git repository.
bbende pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/master by this push:
new 1678531 NIFI-7025: Initial commit adding Kerberos Password feature for Hive components Kerberos Password property should not support EL, this includes a change to KerberosProperties which is also used by the HDFS processors (AbstractHadoopProcessor) Added wiring in a KerberosContext to a TestRunner's MockProcessorInitializationContext Removed synchronization blocks around KerberosUser.checkTGTAndRelogin, since that method is already synchronized Updated AbstractHadoopProcessor t [...]
1678531 is described below
commit 167853163884a20c27cd44b38679918220774411
Author: jstorck <jt...@gmail.com>
AuthorDate: Fri Feb 28 01:18:06 2020 -0500
NIFI-7025: Initial commit adding Kerberos Password feature for Hive components
Kerberos Password property should not support EL, this includes a change to KerberosProperties which is also used by the HDFS processors (AbstractHadoopProcessor)
Added wiring in a KerberosContext to a TestRunner's MockProcessorInitializationContext
Removed synchronization blocks around KerberosUser.checkTGTAndRelogin, since that method is already synchronized
Updated AbstractHadoopProcessor to have a boolean accessor method to determine if explicit keytab configuration is allowed
Removed synchronization block from HiveConnectionPool's getConnection method (in Hive, Hive_1_1, Hive3 modules), since new TGT ticket acquisition is handled by the KerberosUser implementation. If UGI is used to relogin, synchronization is handled internally by UGI.
Added Kerberos Principal and Kerberos Password properties to Hive, Hive_1_1, and Hive3 components
Hive, Hive_1_1, and Hive3 components now use KerberosUser implementations to authenticate with a KDC
Updated handling of the NIFI_ALLOW_EXPLICIT_KEYTAB environment variable in Hive and Hive3 components. An accessor method has been added that uses Boolean.parseBoolean, which returns true if the environment variable is set to true, and false otherwise (including when the environment variable is unset).
Addressing PR feedback
Addressing PR feedback
This closes #4102.
---
.../nifi/security/krb/AbstractKerberosUser.java | 4 +-
.../nifi/security/krb/KeytabConfiguration.java | 4 +
.../util/MockProcessorInitializationContext.java | 15 +++-
.../nifi/util/StandardProcessorTestRunner.java | 15 +++-
.../java/org/apache/nifi/util/TestRunners.java | 22 +++++
.../org/apache/nifi/hadoop/KerberosProperties.java | 3 +-
.../java/org/apache/nifi/hadoop/SecurityUtil.java | 9 ++
.../processors/hadoop/AbstractHadoopProcessor.java | 29 +++----
.../hadoop/AbstractHadoopProcessorSpec.groovy | 30 +++----
.../processors/hadoop/SimpleHadoopProcessor.java | 10 +++
.../apache/nifi/dbcp/hive/HiveConnectionPool.java | 83 ++++++++++++------
.../nifi/processors/hive/PutHiveStreaming.java | 65 ++++++++++++---
.../apache/nifi/util/hive/HiveConfigurator.java | 24 +++++-
.../nifi/processors/hive/TestPutHiveStreaming.java | 24 +++++-
.../apache/nifi/dbcp/hive/Hive3ConnectionPool.java | 95 ++++++++++++++-------
.../nifi/processors/hive/PutHive3Streaming.java | 65 +++++++++++++--
.../apache/nifi/util/hive/HiveConfigurator.java | 24 +++++-
.../processors/hive/TestPutHive3Streaming.java | 10 ++-
.../nifi/dbcp/hive/Hive_1_1ConnectionPool.java | 97 +++++++++++++++++-----
.../apache/nifi/util/hive/HiveConfigurator.java | 24 +++++-
20 files changed, 505 insertions(+), 147 deletions(-)
diff --git a/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/AbstractKerberosUser.java b/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/AbstractKerberosUser.java
index 6764497..5278618 100644
--- a/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/AbstractKerberosUser.java
+++ b/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/AbstractKerberosUser.java
@@ -82,7 +82,9 @@ public abstract class AbstractKerberosUser implements KerberosUser {
loggedIn.set(true);
LOGGER.debug("Successful login for {}", new Object[]{principal});
} catch (LoginException le) {
- throw new LoginException("Unable to login with " + principal + " due to: " + le.getMessage());
+ LoginException loginException = new LoginException("Unable to login with " + principal + " due to: " + le.getMessage());
+ loginException.setStackTrace(le.getStackTrace());
+ throw loginException;
}
}
diff --git a/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/KeytabConfiguration.java b/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/KeytabConfiguration.java
index 24af9b2..a038c1d 100644
--- a/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/KeytabConfiguration.java
+++ b/nifi-commons/nifi-security-utils/src/main/java/org/apache/nifi/security/krb/KeytabConfiguration.java
@@ -17,6 +17,8 @@
package org.apache.nifi.security.krb;
import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.security.auth.login.AppConfigurationEntry;
import javax.security.auth.login.Configuration;
@@ -27,6 +29,7 @@ import java.util.Map;
* Custom JAAS Configuration object for a provided principal and keytab.
*/
public class KeytabConfiguration extends Configuration {
+ private static final Logger LOGGER = LoggerFactory.getLogger(KeytabConfiguration.class);
private final String principal;
private final String keytabFile;
@@ -63,6 +66,7 @@ public class KeytabConfiguration extends Configuration {
final String krbLoginModuleName = ConfigurationUtil.IS_IBM
? ConfigurationUtil.IBM_KRB5_LOGIN_MODULE : ConfigurationUtil.SUN_KRB5_LOGIN_MODULE;
+ LOGGER.debug("krbLoginModuleName: {}, configuration options: {}", krbLoginModuleName, options);
this.kerberosKeytabConfigEntry = new AppConfigurationEntry(
krbLoginModuleName, AppConfigurationEntry.LoginModuleControlFlag.REQUIRED, options);
}
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
index 6a26371..d48fc3d 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockProcessorInitializationContext.java
@@ -23,6 +23,7 @@ import java.util.UUID;
import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.ControllerServiceLookup;
import org.apache.nifi.controller.NodeTypeProvider;
+import org.apache.nifi.kerberos.KerberosContext;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.ProcessorInitializationContext;
@@ -31,15 +32,21 @@ public class MockProcessorInitializationContext implements ProcessorInitializati
private final MockComponentLog logger;
private final String processorId;
private final MockProcessContext context;
+ private final KerberosContext kerberosContext;
public MockProcessorInitializationContext(final Processor processor, final MockProcessContext context) {
- this(processor, context, null);
+ this(processor, context, null, null);
}
public MockProcessorInitializationContext(final Processor processor, final MockProcessContext context, final MockComponentLog logger) {
+ this(processor, context, logger, null);
+ }
+
+ public MockProcessorInitializationContext(final Processor processor, final MockProcessContext context, final MockComponentLog logger, KerberosContext kerberosContext) {
processorId = UUID.randomUUID().toString();
this.logger = logger == null ? new MockComponentLog(processorId, processor) : logger;
this.context = context;
+ this.kerberosContext = kerberosContext;
}
@Override
@@ -94,16 +101,16 @@ public class MockProcessorInitializationContext implements ProcessorInitializati
@Override
public String getKerberosServicePrincipal() {
- return null; //this needs to be wired in.
+ return kerberosContext != null ? kerberosContext.getKerberosServicePrincipal() : null;
}
@Override
public File getKerberosServiceKeytab() {
- return null; //this needs to be wired in.
+ return kerberosContext != null ? kerberosContext.getKerberosServiceKeytab() : null;
}
@Override
public File getKerberosConfigurationFile() {
- return null; //this needs to be wired in.
+ return kerberosContext != null ? kerberosContext.getKerberosConfigurationFile() : null;
}
}
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
index d995f8e..40010c6 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/StandardProcessorTestRunner.java
@@ -64,6 +64,7 @@ import org.apache.nifi.controller.ControllerService;
import org.apache.nifi.controller.queue.QueueSize;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.kerberos.KerberosContext;
import org.apache.nifi.processor.ProcessSessionFactory;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
@@ -77,6 +78,7 @@ public class StandardProcessorTestRunner implements TestRunner {
private final Processor processor;
private final MockProcessContext context;
+ private final KerberosContext kerberosContext;
private final MockFlowFileQueue flowFileQueue;
private final SharedSessionState sharedState;
private final AtomicLong idGenerator;
@@ -99,10 +101,18 @@ public class StandardProcessorTestRunner implements TestRunner {
}
StandardProcessorTestRunner(final Processor processor, String processorName) {
- this(processor, processorName, null);
+ this(processor, processorName, null, null);
+ }
+
+ StandardProcessorTestRunner(final Processor processor, String processorName, KerberosContext kerberosContext) {
+ this(processor, processorName, null, kerberosContext);
}
StandardProcessorTestRunner(final Processor processor, String processorName, MockComponentLog logger) {
+ this(processor, processorName, logger, null);
+ }
+
+ StandardProcessorTestRunner(final Processor processor, String processorName, MockComponentLog logger, KerberosContext kerberosContext) {
this.processor = processor;
this.idGenerator = new AtomicLong(0L);
this.sharedState = new SharedSessionState(processor, idGenerator);
@@ -111,8 +121,9 @@ public class StandardProcessorTestRunner implements TestRunner {
this.processorStateManager = new MockStateManager(processor);
this.variableRegistry = new MockVariableRegistry();
this.context = new MockProcessContext(processor, processorName, processorStateManager, variableRegistry);
+ this.kerberosContext = kerberosContext;
- final MockProcessorInitializationContext mockInitContext = new MockProcessorInitializationContext(processor, context, logger);
+ final MockProcessorInitializationContext mockInitContext = new MockProcessorInitializationContext(processor, context, logger, kerberosContext);
processor.initialize(mockInitContext);
this.logger = mockInitContext.getLogger();
diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
index 9792473..8370e88 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/TestRunners.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.util;
+import org.apache.nifi.kerberos.KerberosContext;
import org.apache.nifi.processor.Processor;
public class TestRunners {
@@ -31,6 +32,16 @@ public class TestRunners {
}
/**
+ * Returns a {@code TestRunner} for the given {@code Processor} which uses the given {@code KerberosContext}.
+ * @param processor the {@code Processor} under test
+ * @param kerberosContext the {@code KerberosContext} used during the test
+ * @return
+ */
+ public static TestRunner newTestRunner(final Processor processor, KerberosContext kerberosContext) {
+ return newTestRunner(processor,processor.getClass().getName(), kerberosContext);
+ }
+
+ /**
* Returns a {@code TestRunner} for the given {@code Processor}.
* The processor name available from {@code TestRunner.getProcessContext().getName()} will have the default name of {@code processor.getClass().getName()}
* @param processor the {@code Processor} under test
@@ -53,6 +64,17 @@ public class TestRunners {
}
/**
+ * Returns a {@code TestRunner} for the given {@code Processor} and {@code KerberosContext}.
+ * @param processor the {@code Processor} under test
+ * @param name the name to give the {@code Processor}
+ * @param kerberosContext the {@code KerberosContext} used during the test
+ * @return a {@code TestRunner}
+ */
+ public static TestRunner newTestRunner(final Processor processor, String name, KerberosContext kerberosContext) {
+ return new StandardProcessorTestRunner(processor, name, kerberosContext);
+ }
+
+ /**
* Returns a {@code TestRunner} for the given {@code Processor}.
* The processor name available from {@code TestRunner.getProcessContext().getName()} will be the passed name.
* @param processor the {@code Processor} under test
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/KerberosProperties.java b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/KerberosProperties.java
index ba5e9ec..5977b6c 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/KerberosProperties.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/KerberosProperties.java
@@ -104,8 +104,7 @@ public class KerberosProperties {
.name("Kerberos Password")
.required(false)
.description("Kerberos password associated with the principal.")
- .addValidator(StandardValidators.ATTRIBUTE_EXPRESSION_LANGUAGE_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.sensitive(true)
.build();
}
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/SecurityUtil.java b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/SecurityUtil.java
index 6a8f079..eeabaf0 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/SecurityUtil.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/hadoop/SecurityUtil.java
@@ -78,6 +78,15 @@ public class SecurityUtil {
return UserGroupInformation.getCurrentUser();
}
+ /**
+ * Authenticates a {@link KerberosUser} and acquires a {@link UserGroupInformation} instance using {@link UserGroupInformation#getUGIFromSubject(Subject)}.
+ * The {@link UserGroupInformation} will use the given {@link Configuration}.
+ *
+ * @param config The Configuration to apply to the acquired UserGroupInformation instance
+ * @param kerberosUser The KerberosUser to authenticate
+ * @return A UserGroupInformation instance created using the Subject of the given KerberosUser
+ * @throws IOException if authentication fails
+ */
public static synchronized UserGroupInformation getUgiForKerberosUser(final Configuration config, final KerberosUser kerberosUser) throws IOException {
UserGroupInformation.setConfiguration(config);
try {
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
index 69ea716..7a59170 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java
@@ -184,12 +184,11 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
final String configResources = validationContext.getProperty(HADOOP_CONFIGURATION_RESOURCES).evaluateAttributeExpressions().getValue();
final String explicitPrincipal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
final String explicitKeytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
- final String explicitPassword = validationContext.getProperty(kerberosProperties.getKerberosPassword()).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = validationContext.getProperty(kerberosProperties.getKerberosPassword()).getValue();
final KerberosCredentialsService credentialsService = validationContext.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
final String resolvedKeytab;
- final String resolvedPassword;
if (credentialsService == null) {
resolvedPrincipal = explicitPrincipal;
resolvedKeytab = explicitKeytab;
@@ -197,7 +196,6 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
resolvedPrincipal = credentialsService.getPrincipal();
resolvedKeytab = credentialsService.getKeytab();
}
- resolvedPassword = explicitPassword;
final List<ValidationResult> results = new ArrayList<>();
@@ -220,7 +218,7 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
final Configuration conf = resources.getConfiguration();
results.addAll(KerberosProperties.validatePrincipalWithKeytabOrPassword(
- this.getClass().getSimpleName(), conf, resolvedPrincipal, resolvedKeytab, resolvedPassword, getLogger()));
+ this.getClass().getSimpleName(), conf, resolvedPrincipal, resolvedKeytab, explicitPassword, getLogger()));
} catch (final IOException e) {
results.add(new ValidationResult.Builder()
@@ -238,8 +236,7 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
.build());
}
- final String allowExplicitKeytabVariable = getAllowExplicitKeytabEnvironmentVariable();
- if ("false".equalsIgnoreCase(allowExplicitKeytabVariable) && explicitKeytab != null) {
+ if (!isAllowExplicitKeytab() && explicitKeytab != null) {
results.add(new ValidationResult.Builder()
.subject("Kerberos Credentials")
.valid(false)
@@ -390,7 +387,7 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
if (SecurityUtil.isSecurityEnabled(config)) {
String principal = context.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
String keyTab = context.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
- String password = context.getProperty(kerberosProperties.getKerberosPassword()).evaluateAttributeExpressions().getValue();
+ String password = context.getProperty(kerberosProperties.getKerberosPassword()).getValue();
// If the Kerberos Credentials Service is specified, we need to use its configuration, not the explicit properties for principal/keytab.
// The customValidate method ensures that only one can be set, so we know that the principal & keytab above are null.
@@ -545,15 +542,13 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
if (hdfsResources.get().getKerberosUser() != null) {
// if there's a KerberosUser associated with this UGI, check the TGT and relogin if it is close to expiring
KerberosUser kerberosUser = hdfsResources.get().getKerberosUser();
- synchronized (kerberosUser) {
- getLogger().debug("kerberosUser is " + kerberosUser);
- try {
- getLogger().debug("checking TGT on kerberosUser [{}]" + kerberosUser);
- kerberosUser.checkTGTAndRelogin();
- } catch (LoginException e) {
- throw new ProcessException("Unable to relogin with kerberos credentials for " + kerberosUser.getPrincipal(), e);
+ getLogger().debug("kerberosUser is " + kerberosUser);
+ try {
+ getLogger().debug("checking TGT on kerberosUser " + kerberosUser);
+ kerberosUser.checkTGTAndRelogin();
+ } catch (LoginException e) {
+ throw new ProcessException("Unable to relogin with kerberos credentials for " + kerberosUser.getPrincipal(), e);
}
- }
} else {
getLogger().debug("kerberosUser was null, will not refresh TGT with KerberosUser");
}
@@ -563,8 +558,8 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor {
/*
* Overridable by subclasses in the same package, mainly intended for testing purposes to allow verification without having to set environment variables.
*/
- String getAllowExplicitKeytabEnvironmentVariable() {
- return System.getenv(ALLOW_EXPLICIT_KEYTAB);
+ boolean isAllowExplicitKeytab() {
+ return Boolean.parseBoolean(System.getenv(ALLOW_EXPLICIT_KEYTAB));
}
static protected class HdfsResources {
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/test/groovy/org/apache/nifi/processors/hadoop/AbstractHadoopProcessorSpec.groovy b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/test/groovy/org/apache/nifi/processors/hadoop/AbstractHadoopProcessorSpec.groovy
index 9b4faf5..49a835f 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/test/groovy/org/apache/nifi/processors/hadoop/AbstractHadoopProcessorSpec.groovy
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-hadoop-utils/src/test/groovy/org/apache/nifi/processors/hadoop/AbstractHadoopProcessorSpec.groovy
@@ -74,22 +74,22 @@ class AbstractHadoopProcessorSpec extends Specification {
actualValidationErrors.size() == expectedValidationErrorCount
where:
- testName | configuredPrincipal | configuredKeytab | configuredPassword | allowExplicitKeytab | configuredKeytabCredentialsService | configuredKeytabCredentialsServicePrincipal | configuredKeytabCredentialsServiceKeytab || expectedValidationErrorCount
- "success case 1" | "principal" | "keytab" | null | "true" | false | null | null || 0
- "success case 2" | "principal" | null | "password" | "true" | false | null | null || 0
- "success case 3" | "principal" | null | "password" | "false" | false | null | null || 0
- "success case 4" | null | null | null | "true" | true | "principal" | "keytab" || 0
- "success case 5" | null | null | null | "false" | true | "principal" | "keytab" || 0
+ testName | configuredPrincipal | configuredKeytab | configuredPassword | allowExplicitKeytab | configuredKeytabCredentialsService | configuredKeytabCredentialsServicePrincipal | configuredKeytabCredentialsServiceKeytab || expectedValidationErrorCount
+ "success case 1" | "principal" | "keytab" | null | true | false | null | null || 0
+ "success case 2" | "principal" | null | "password" | true | false | null | null || 0
+ "success case 3" | "principal" | null | "password" | false | false | null | null || 0
+ "success case 4" | null | null | null | true | true | "principal" | "keytab" || 0
+ "success case 5" | null | null | null | false | true | "principal" | "keytab" || 0
// do not allow explicit keytab, but provide one anyway; validation fails
- "failure case 1" | "principal" | "keytab" | null | "false" | false | null | null || 1
- "failure case 2" | null | "keytab" | null | "false" | false | null | null || 2
+ "failure case 1" | "principal" | "keytab" | null | false | false | null | null || 1
+ "failure case 2" | null | "keytab" | null | false | false | null | null || 2
// keytab credentials service is provided, but explicit properties for principal, password, or keytab are also provided; validation fails
- "failure case 3" | "principal" | null | null | "true" | true | "principal" | "keytab" || 1
- "failure case 4" | null | "keytab" | null | "true" | true | "principal" | "keytab" || 1
- "failure case 5" | null | null | "password" | "true" | true | "principal" | "keytab" || 2
- "failure case 6" | "principal" | null | null | "false" | true | "principal" | "keytab" || 1
- "failure case 7" | null | "keytab" | null | "false" | true | "principal" | "keytab" || 2
- "failure case 8" | null | null | "password" | "false" | true | "principal" | "keytab" || 2
+ "failure case 3" | "principal" | null | null | true | true | "principal" | "keytab" || 1
+ "failure case 4" | null | "keytab" | null | true | true | "principal" | "keytab" || 1
+ "failure case 5" | null | null | "password" | true | true | "principal" | "keytab" || 2
+ "failure case 6" | "principal" | null | null | false | true | "principal" | "keytab" || 1
+ "failure case 7" | null | "keytab" | null | false | true | "principal" | "keytab" || 2
+ "failure case 8" | null | null | "password" | false | true | "principal" | "keytab" || 2
}
private class TestAbstractHadoopProcessor extends AbstractHadoopProcessor {
@@ -105,7 +105,7 @@ class AbstractHadoopProcessorSpec extends Specification {
}
@Override
- String getAllowExplicitKeytabEnvironmentVariable() {
+ boolean isAllowExplicitKeytab() {
allowExplicitKeytab
}
diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/SimpleHadoopProcessor.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/SimpleHadoopProcessor.java
index 4aa8e54..f8622da 100644
--- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/SimpleHadoopProcessor.java
+++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/SimpleHadoopProcessor.java
@@ -26,9 +26,15 @@ import java.io.File;
public class SimpleHadoopProcessor extends AbstractHadoopProcessor {
private KerberosProperties testKerberosProperties;
+ private boolean allowExplicitKeytab;
public SimpleHadoopProcessor(KerberosProperties kerberosProperties) {
+ this(kerberosProperties, true);
+ }
+
+ public SimpleHadoopProcessor(KerberosProperties kerberosProperties, boolean allowExplicitKeytab) {
this.testKerberosProperties = kerberosProperties;
+ this.allowExplicitKeytab = allowExplicitKeytab;
}
@Override
@@ -40,4 +46,8 @@ public class SimpleHadoopProcessor extends AbstractHadoopProcessor {
return testKerberosProperties;
}
+ @Override
+ boolean isAllowExplicitKeytab() {
+ return allowExplicitKeytab;
+ }
}
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java
index 378799e..f84af6f 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java
@@ -39,6 +39,9 @@ import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.security.krb.KerberosKeytabUser;
+import org.apache.nifi.security.krb.KerberosPasswordUser;
+import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.util.hive.AuthenticationFailedException;
import org.apache.nifi.util.hive.HiveConfigurator;
import org.apache.nifi.util.hive.HiveUtils;
@@ -59,6 +62,8 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.controller.ControllerServiceInitializationContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
+import javax.security.auth.login.LoginException;
+
/**
* Implementation for Database Connection Pooling Service used for Apache Hive
* connections. Apache DBCP is used for connection pooling functionality.
@@ -165,6 +170,7 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv
private volatile HiveConfigurator hiveConfigurator = new HiveConfigurator();
private volatile UserGroupInformation ugi;
+ private final AtomicReference<KerberosUser> kerberosUserReference = new AtomicReference<>();
private volatile File kerberosConfigFile = null;
private volatile KerberosProperties kerberosProperties;
@@ -184,6 +190,7 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv
kerberosProperties = new KerberosProperties(kerberosConfigFile);
props.add(kerberosProperties.getKerberosPrincipal());
props.add(kerberosProperties.getKerberosKeytab());
+ props.add(kerberosProperties.getKerberosPassword());
properties = props;
}
@@ -201,37 +208,37 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv
if (confFileProvided) {
final String explicitPrincipal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
final String explicitKeytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = validationContext.getProperty(kerberosProperties.getKerberosPassword()).getValue();
final KerberosCredentialsService credentialsService = validationContext.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
final String resolvedKeytab;
- if (credentialsService == null) {
- resolvedPrincipal = explicitPrincipal;
- resolvedKeytab = explicitKeytab;
- } else {
+ if (credentialsService != null) {
resolvedPrincipal = credentialsService.getPrincipal();
resolvedKeytab = credentialsService.getKeytab();
+ } else {
+ resolvedPrincipal = explicitPrincipal;
+ resolvedKeytab = explicitKeytab;
}
final String configFiles = validationContext.getProperty(HIVE_CONFIGURATION_RESOURCES).evaluateAttributeExpressions().getValue();
- problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, validationResourceHolder, getLogger()));
+ problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, explicitPassword, validationResourceHolder, getLogger()));
- if (credentialsService != null && (explicitPrincipal != null || explicitKeytab != null)) {
+ if (credentialsService != null && (explicitPrincipal != null || explicitKeytab != null || explicitPassword != null)) {
problems.add(new ValidationResult.Builder()
.subject("Kerberos Credentials")
.valid(false)
- .explanation("Cannot specify both a Kerberos Credentials Service and a principal/keytab")
+ .explanation("Cannot specify a Kerberos Credentials Service while also specifying a Kerberos Principal, Kerberos Keytab, or Kerberos Password")
.build());
}
- final String allowExplicitKeytabVariable = System.getenv(ALLOW_EXPLICIT_KEYTAB);
- if ("false".equalsIgnoreCase(allowExplicitKeytabVariable) && (explicitPrincipal != null || explicitKeytab != null)) {
+ if (!isAllowExplicitKeytab() && explicitKeytab != null) {
problems.add(new ValidationResult.Builder()
.subject("Kerberos Credentials")
.valid(false)
- .explanation("The '" + ALLOW_EXPLICIT_KEYTAB + "' system environment variable is configured to forbid explicitly configuring principal/keytab in processors. "
- + "The Kerberos Credentials Service should be used instead of setting the Kerberos Keytab or Kerberos Principal property.")
+ .explanation("The '" + ALLOW_EXPLICIT_KEYTAB + "' system environment variable is configured to forbid explicitly configuring Kerberos Keytab in processors. "
+ + "The Kerberos Credentials Service should be used instead of setting the Kerberos Keytab or Kerberos Principal property.")
.build());
}
}
@@ -293,28 +300,37 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv
if (SecurityUtil.isSecurityEnabled(hiveConfig)) {
final String explicitPrincipal = context.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
final String explicitKeytab = context.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = context.getProperty(kerberosProperties.getKerberosPassword()).getValue();
final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
final String resolvedKeytab;
- if (credentialsService == null) {
- resolvedPrincipal = explicitPrincipal;
- resolvedKeytab = explicitKeytab;
- } else {
+ if (credentialsService != null) {
resolvedPrincipal = credentialsService.getPrincipal();
resolvedKeytab = credentialsService.getKeytab();
+ } else {
+ resolvedPrincipal = explicitPrincipal;
+ resolvedKeytab = explicitKeytab;
}
- log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ if (resolvedKeytab != null) {
+ kerberosUserReference.set(new KerberosKeytabUser(resolvedPrincipal, resolvedKeytab));
+ log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ } else if (explicitPassword != null) {
+ kerberosUserReference.set(new KerberosPasswordUser(resolvedPrincipal, explicitPassword));
+ log.info("Hive Security Enabled, logging in as principal {} with password", new Object[] {resolvedPrincipal});
+ } else {
+ throw new InitializationException("Unable to authenticate with Kerberos, no keytab or password was provided");
+ }
try {
- ugi = hiveConfigurator.authenticate(hiveConfig, resolvedPrincipal, resolvedKeytab);
+ ugi = hiveConfigurator.authenticate(hiveConfig, kerberosUserReference.get());
} catch (AuthenticationFailedException ae) {
log.error(ae.getMessage(), ae);
throw new InitializationException(ae);
}
- getLogger().info("Successfully logged in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ getLogger().info("Successfully logged in as principal " + resolvedPrincipal);
}
final String user = context.getProperty(DB_USER).evaluateAttributeExpressions().getValue();
@@ -356,13 +372,24 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv
public Connection getConnection() throws ProcessException {
try {
if (ugi != null) {
- synchronized(this) {
- /*
- * Make sure that only one thread can request that the UGI relogin at a time. This
- * explicit relogin attempt is necessary due to the Hive client/thrift not implicitly handling
- * the acquisition of a new TGT after the current one has expired.
- * https://issues.apache.org/jira/browse/NIFI-5134
- */
+ /*
+ * Explicitly check the TGT and relogin if necessary with the KerberosUser instance. No synchronization
+ * is necessary in the client code, since AbstractKerberosUser's checkTGTAndRelogin method is synchronized.
+ */
+ getLogger().trace("getting UGI instance");
+ if (kerberosUserReference.get() != null) {
+ // if there's a KerberosUser associated with this UGI, check the TGT and relogin if it is close to expiring
+ KerberosUser kerberosUser = kerberosUserReference.get();
+ getLogger().debug("kerberosUser is " + kerberosUser);
+ try {
+ getLogger().debug("checking TGT on kerberosUser [{}]", new Object[]{kerberosUser});
+ kerberosUser.checkTGTAndRelogin();
+ } catch (LoginException e) {
+ throw new ProcessException("Unable to relogin with kerberos credentials for " + kerberosUser.getPrincipal(), e);
+ }
+ } else {
+ getLogger().debug("kerberosUser was null, will not refresh TGT with KerberosUser");
+ // no synchronization is needed for UserGroupInformation.checkTGTAndReloginFromKeytab; UGI handles the synchronization internally
ugi.checkTGTAndReloginFromKeytab();
}
try {
@@ -395,4 +422,10 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv
return connectionUrl;
}
+ /*
+ * Overridable by subclasses in the same package, mainly intended for testing purposes to allow verification without having to set environment variables.
+ */
+ boolean isAllowExplicitKeytab() {
+ return Boolean.parseBoolean(System.getenv(ALLOW_EXPLICIT_KEYTAB));
+ }
}
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java
index 1627159..0ef4211 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java
@@ -62,6 +62,9 @@ import org.apache.nifi.processor.util.pattern.ErrorTypes;
import org.apache.nifi.processor.util.pattern.ExceptionHandler;
import org.apache.nifi.processor.util.pattern.RollbackOnFailure;
import org.apache.nifi.processor.util.pattern.RoutingResult;
+import org.apache.nifi.security.krb.KerberosKeytabUser;
+import org.apache.nifi.security.krb.KerberosPasswordUser;
+import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.util.hive.AuthenticationFailedException;
import org.apache.nifi.util.hive.HiveConfigurator;
import org.apache.nifi.util.hive.HiveOptions;
@@ -70,6 +73,7 @@ import org.apache.nifi.util.hive.HiveWriter;
import org.xerial.snappy.Snappy;
import org.apache.nifi.util.hive.ValidationResources;
+import javax.security.auth.login.LoginException;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
@@ -318,6 +322,7 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
protected volatile HiveConfigurator hiveConfigurator = new HiveConfigurator();
protected volatile UserGroupInformation ugi;
+ final protected AtomicReference<KerberosUser> kerberosUserReference = new AtomicReference<>();
protected volatile HiveConf hiveConfig;
protected final AtomicBoolean sendHeartBeat = new AtomicBoolean(false);
@@ -353,6 +358,7 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
kerberosProperties = new KerberosProperties(kerberosConfigFile);
props.add(kerberosProperties.getKerberosPrincipal());
props.add(kerberosProperties.getKerberosKeytab());
+ props.add(kerberosProperties.getKerberosPassword());
propertyDescriptors = Collections.unmodifiableList(props);
Set<Relationship> _relationships = new HashSet<>();
@@ -382,6 +388,7 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
if (confFileProvided) {
final String explicitPrincipal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
final String explicitKeytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = validationContext.getProperty(kerberosProperties.getKerberosPassword()).getValue();
final KerberosCredentialsService credentialsService = validationContext.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
@@ -396,23 +403,22 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
final String configFiles = validationContext.getProperty(HIVE_CONFIGURATION_RESOURCES).evaluateAttributeExpressions().getValue();
- problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, validationResourceHolder, getLogger()));
+ problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, explicitPassword, validationResourceHolder, getLogger()));
- if (credentialsService != null && (explicitPrincipal != null || explicitKeytab != null)) {
+ if (credentialsService != null && (explicitPrincipal != null || explicitKeytab != null || explicitPassword != null)) {
problems.add(new ValidationResult.Builder()
.subject("Kerberos Credentials")
.valid(false)
- .explanation("Cannot specify both a Kerberos Credentials Service and a principal/keytab")
+ .explanation("Cannot specify a Kerberos Credentials Service while also specifying a Kerberos Principal, Kerberos Keytab, or Kerberos Password")
.build());
}
- final String allowExplicitKeytabVariable = System.getenv(ALLOW_EXPLICIT_KEYTAB);
- if ("false".equalsIgnoreCase(allowExplicitKeytabVariable) && (explicitPrincipal != null || explicitKeytab != null)) {
+ if (!isAllowExplicitKeytab() && explicitKeytab != null) {
problems.add(new ValidationResult.Builder()
.subject("Kerberos Credentials")
.valid(false)
- .explanation("The '" + ALLOW_EXPLICIT_KEYTAB + "' system environment variable is configured to forbid explicitly configuring principal/keytab in processors. "
- + "The Kerberos Credentials Service should be used instead of setting the Kerberos Keytab or Kerberos Principal property.")
+ .explanation("The '" + ALLOW_EXPLICIT_KEYTAB + "' system environment variable is configured to forbid explicitly configuring Kerberos Keytab in processors. "
+ + "The Kerberos Credentials Service should be used instead of setting the Kerberos Keytab or Kerberos Principal property.")
.build());
}
}
@@ -446,6 +452,7 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
if (SecurityUtil.isSecurityEnabled(hiveConfig)) {
final String explicitPrincipal = context.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
final String explicitKeytab = context.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = context.getProperty(kerberosProperties.getKerberosPassword()).getValue();
final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
@@ -458,16 +465,26 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
resolvedKeytab = credentialsService.getKeytab();
}
- log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ if (resolvedKeytab != null) {
+ kerberosUserReference.set(new KerberosKeytabUser(resolvedPrincipal, resolvedKeytab));
+ log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ } else if (explicitPassword != null) {
+ kerberosUserReference.set(new KerberosPasswordUser(resolvedPrincipal, explicitPassword));
+ log.info("Hive Security Enabled, logging in as principal {} with password", new Object[] {resolvedPrincipal});
+ } else {
+ throw new ProcessException("Unable to authenticate with Kerberos, no keytab or password was provided");
+ }
+
try {
- ugi = hiveConfigurator.authenticate(hiveConfig, resolvedPrincipal, resolvedKeytab);
+ ugi = hiveConfigurator.authenticate(hiveConfig, kerberosUserReference.get());
} catch (AuthenticationFailedException ae) {
throw new ProcessException("Kerberos authentication failed for Hive Streaming", ae);
}
- log.info("Successfully logged in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ log.info("Successfully logged in as principal " + resolvedPrincipal);
} else {
ugi = null;
+ kerberosUserReference.set(null);
}
callTimeout = context.getProperty(CALL_TIMEOUT).evaluateAttributeExpressions().asInteger() * 1000; // milliseconds
@@ -967,6 +984,7 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
}
ugi = null;
+ kerberosUserReference.set(null);
}
private void setupHeartBeatTimer(int heartbeatInterval) {
@@ -1048,7 +1066,7 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
HiveWriter writer = writers.get(endPoint);
if (writer == null) {
log.debug("Creating Writer to Hive end point : " + endPoint);
- writer = makeHiveWriter(endPoint, callTimeoutPool, ugi, options);
+ writer = makeHiveWriter(endPoint, callTimeoutPool, getUgi(), options);
if (writers.size() > (options.getMaxOpenConnections() - 1)) {
log.info("cached HiveEndPoint size {} exceeded maxOpenConnections {} ", new Object[]{writers.size(), options.getMaxOpenConnections()});
int retired = retireIdleWriters(writers, options.getIdleTimeout());
@@ -1144,6 +1162,31 @@ public class PutHiveStreaming extends AbstractSessionFactoryProcessor {
return kerberosProperties;
}
+ UserGroupInformation getUgi() {
+ getLogger().trace("getting UGI instance");
+ if (kerberosUserReference.get() != null) {
+ // if there's a KerberosUser associated with this UGI, check the TGT and relogin if it is close to expiring
+ KerberosUser kerberosUser = kerberosUserReference.get();
+ getLogger().debug("kerberosUser is " + kerberosUser);
+ try {
+ getLogger().debug("checking TGT on kerberosUser [{}]", new Object[] {kerberosUser});
+ kerberosUser.checkTGTAndRelogin();
+ } catch (LoginException e) {
+ throw new ProcessException("Unable to relogin with kerberos credentials for " + kerberosUser.getPrincipal(), e);
+ }
+ } else {
+ getLogger().debug("kerberosUser was null, will not refresh TGT with KerberosUser");
+ }
+ return ugi;
+ }
+
+ /*
+ * Overridable by subclasses in the same package, mainly intended for testing purposes to allow verification without having to set environment variables.
+ */
+ boolean isAllowExplicitKeytab() {
+ return Boolean.parseBoolean(System.getenv(ALLOW_EXPLICIT_KEYTAB));
+ }
+
protected class HiveStreamingRecord {
private List<String> partitionValues;
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
index 5fed81a..b212207 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
@@ -26,6 +26,7 @@ import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.hadoop.KerberosProperties;
import org.apache.nifi.hadoop.SecurityUtil;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.security.krb.KerberosUser;
import java.io.IOException;
import java.util.ArrayList;
@@ -35,7 +36,8 @@ import java.util.concurrent.atomic.AtomicReference;
public class HiveConfigurator {
- public Collection<ValidationResult> validate(String configFiles, String principal, String keyTab, AtomicReference<ValidationResources> validationResourceHolder, ComponentLog log) {
+ public Collection<ValidationResult> validate(String configFiles, String principal, String keyTab, String password,
+ AtomicReference<ValidationResources> validationResourceHolder, ComponentLog log) {
final List<ValidationResult> problems = new ArrayList<>();
ValidationResources resources = validationResourceHolder.get();
@@ -50,7 +52,7 @@ public class HiveConfigurator {
final Configuration hiveConfig = resources.getConfiguration();
- problems.addAll(KerberosProperties.validatePrincipalWithKeytabOrPassword(this.getClass().getSimpleName(), hiveConfig, principal, keyTab, null, log));
+ problems.addAll(KerberosProperties.validatePrincipalWithKeytabOrPassword(this.getClass().getSimpleName(), hiveConfig, principal, keyTab, password, log));
return problems;
}
@@ -75,6 +77,22 @@ public class HiveConfigurator {
}
/**
+ * Acquires a {@link UserGroupInformation} using the given {@link Configuration} and {@link KerberosUser}.
+ * @see SecurityUtil#getUgiForKerberosUser(Configuration, KerberosUser)
+ * @param hiveConfig The Configuration to apply to the acquired UserGroupInformation
+ * @param kerberosUser The KerberosUser to authenticate
+ * @return A UserGroupInformation instance created using the Subject of the given KerberosUser
+ * @throws AuthenticationFailedException if authentication fails
+ */
+ public UserGroupInformation authenticate(final Configuration hiveConfig, KerberosUser kerberosUser) throws AuthenticationFailedException {
+ try {
+ return SecurityUtil.getUgiForKerberosUser(hiveConfig, kerberosUser);
+ } catch (IOException ioe) {
+ throw new AuthenticationFailedException("Kerberos Authentication for Hive failed", ioe);
+ }
+ }
+
+ /**
* As of Apache NiFi 1.5.0, due to changes made to
* {@link SecurityUtil#loginKerberos(Configuration, String, String)}, which is used by this
* class to authenticate a principal with Kerberos, Hive controller services no longer
@@ -91,7 +109,9 @@ public class HiveConfigurator {
* authentication attempts that would leave the Hive controller service in an unrecoverable state.
*
* @see SecurityUtil#loginKerberos(Configuration, String, String)
+ * @deprecated Use {@link SecurityUtil#getUgiForKerberosUser(Configuration, KerberosUser)}
*/
+ @Deprecated
public UserGroupInformation authenticate(final Configuration hiveConfig, String principal, String keyTab) throws AuthenticationFailedException {
UserGroupInformation ugi;
try {
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java
index ed3461d..3681755 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java
@@ -34,7 +34,10 @@ import org.apache.hive.hcatalog.streaming.StreamingException;
import org.apache.hive.hcatalog.streaming.TransactionBatch;
import org.apache.nifi.hadoop.KerberosProperties;
import org.apache.nifi.hadoop.SecurityUtil;
+import org.apache.nifi.kerberos.KerberosContext;
import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.security.krb.KerberosPasswordUser;
+import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
@@ -67,6 +70,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
@@ -103,7 +107,11 @@ public class TestPutHiveStreaming {
when(hiveConfigurator.getConfigurationFromFiles(AdditionalMatchers.or(anyString(), isNull()))).thenReturn(hiveConf);
processor.hiveConfigurator = hiveConfigurator;
processor.setKerberosProperties(kerberosPropsWithFile);
- runner = TestRunners.newTestRunner(processor);
+ KerberosContext mockKerberosContext = mock(KerberosContext.class);
+ when(mockKerberosContext.getKerberosConfigurationFile()).thenReturn(kerberosPropsWithFile.getKerberosConfigFile());
+ when(mockKerberosContext.getKerberosServiceKeytab()).thenReturn(null);
+ when(mockKerberosContext.getKerberosServicePrincipal()).thenReturn(null);
+ runner = TestRunners.newTestRunner(processor, mockKerberosContext);
}
@Test
@@ -118,23 +126,27 @@ public class TestPutHiveStreaming {
}
@Test
- public void testUgiGetsCleared() {
+ public void testUgiAndKerberosUserGetsCleared() {
runner.setProperty(PutHiveStreaming.METASTORE_URI, "thrift://localhost:9083");
runner.setProperty(PutHiveStreaming.DB_NAME, "default");
runner.setProperty(PutHiveStreaming.TABLE_NAME, "users");
processor.ugi = mock(UserGroupInformation.class);
+ processor.kerberosUserReference.set(new KerberosPasswordUser("user", "password"));
runner.run();
assertNull(processor.ugi);
+ assertNull(processor.kerberosUserReference.get());
}
@Test
public void testUgiGetsSetIfSecure() throws AuthenticationFailedException, IOException {
when(hiveConf.get(SecurityUtil.HADOOP_SECURITY_AUTHENTICATION)).thenReturn(SecurityUtil.KERBEROS);
ugi = mock(UserGroupInformation.class);
- when(hiveConfigurator.authenticate(eq(hiveConf), AdditionalMatchers.or(anyString(), isNull()), AdditionalMatchers.or(anyString(), isNull()))).thenReturn(ugi);
+ when(hiveConfigurator.authenticate(eq(hiveConf), any(KerberosUser.class))).thenReturn(ugi);
runner.setProperty(PutHiveStreaming.METASTORE_URI, "thrift://localhost:9083");
runner.setProperty(PutHiveStreaming.DB_NAME, "default");
runner.setProperty(PutHiveStreaming.TABLE_NAME, "users");
+ runner.setProperty(kerberosPropsWithFile.getKerberosKeytab(), "src/test/resources/fake.keytab");
+ runner.setProperty(kerberosPropsWithFile.getKerberosPrincipal(), "principal");
Map<String, Object> user1 = new HashMap<String, Object>() {
{
put("name", "Joe");
@@ -163,7 +175,7 @@ public class TestPutHiveStreaming {
runner.setProperty(PutHiveStreaming.TABLE_NAME, "users");
runner.setProperty(PutHiveStreaming.HIVE_CONFIGURATION_RESOURCES, "src/test/resources/core-site-security.xml, src/test/resources/hive-site-security.xml");
runner.setProperty(kerberosPropsWithFile.getKerberosPrincipal(), "test@REALM");
- runner.setProperty(kerberosPropsWithFile.getKerberosKeytab(), "src/test/resources/fake.keytab");
+ runner.setProperty(kerberosPropsWithFile.getKerberosKeytab(), "src/test/resources/missing.keytab");
runner.run();
}
@@ -1037,6 +1049,10 @@ public class TestPutHiveStreaming {
this.generateExceptionOnFlushAndClose = generateExceptionOnFlushAndClose;
}
+ @Override
+ UserGroupInformation getUgi() {
+ return ugi;
+ }
}
private class MockHiveWriter extends HiveWriter {
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive3ConnectionPool.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive3ConnectionPool.java
index 01597fa..0a49a40 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive3ConnectionPool.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive3ConnectionPool.java
@@ -43,6 +43,9 @@ import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.security.krb.KerberosKeytabUser;
+import org.apache.nifi.security.krb.KerberosPasswordUser;
+import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.util.hive.AuthenticationFailedException;
import org.apache.nifi.util.hive.HiveConfigurator;
import org.apache.nifi.util.hive.HiveUtils;
@@ -62,6 +65,8 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.controller.ControllerServiceInitializationContext;
+import javax.security.auth.login.LoginException;
+
/**
* Implementation for Database Connection Pooling Service used for Apache Hive
* connections. Apache DBCP is used for connection pooling functionality.
@@ -264,6 +269,7 @@ public class Hive3ConnectionPool extends AbstractControllerService implements Hi
private volatile HiveConfigurator hiveConfigurator = new HiveConfigurator();
private volatile UserGroupInformation ugi;
+ private final AtomicReference<KerberosUser> kerberosUserReference = new AtomicReference<>();
private volatile File kerberosConfigFile = null;
private volatile KerberosProperties kerberosProperties;
@@ -289,6 +295,7 @@ public class Hive3ConnectionPool extends AbstractControllerService implements Hi
kerberosProperties = new KerberosProperties(kerberosConfigFile);
props.add(kerberosProperties.getKerberosPrincipal());
props.add(kerberosProperties.getKerberosKeytab());
+ props.add(kerberosProperties.getKerberosPassword());
properties = props;
}
@@ -306,38 +313,38 @@ public class Hive3ConnectionPool extends AbstractControllerService implements Hi
if (confFileProvided) {
final String explicitPrincipal = validationContext.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
final String explicitKeytab = validationContext.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = validationContext.getProperty(kerberosProperties.getKerberosPassword()).getValue();
final KerberosCredentialsService credentialsService = validationContext.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
final String resolvedKeytab;
- if (credentialsService == null) {
- resolvedPrincipal = explicitPrincipal;
- resolvedKeytab = explicitKeytab;
- } else {
+ if (credentialsService != null) {
resolvedPrincipal = credentialsService.getPrincipal();
resolvedKeytab = credentialsService.getKeytab();
+ } else {
+ resolvedPrincipal = explicitPrincipal;
+ resolvedKeytab = explicitKeytab;
}
final String configFiles = validationContext.getProperty(HIVE_CONFIGURATION_RESOURCES).evaluateAttributeExpressions().getValue();
- problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, validationResourceHolder, getLogger()));
+ problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, explicitPassword, validationResourceHolder, getLogger()));
- if (credentialsService != null && (explicitPrincipal != null || explicitKeytab != null)) {
+ if (credentialsService != null && (explicitPrincipal != null || explicitKeytab != null || explicitPassword != null)) {
problems.add(new ValidationResult.Builder()
- .subject("Kerberos Credentials")
- .valid(false)
- .explanation("Cannot specify both a Kerberos Credentials Service and a principal/keytab")
- .build());
+ .subject("Kerberos Credentials")
+ .valid(false)
+ .explanation("Cannot specify a Kerberos Credentials Service while also specifying a Kerberos Principal, Kerberos Keytab, or Kerberos Password")
+ .build());
}
- final String allowExplicitKeytabVariable = System.getenv(ALLOW_EXPLICIT_KEYTAB);
- if ("false".equalsIgnoreCase(allowExplicitKeytabVariable) && (explicitPrincipal != null || explicitKeytab != null)) {
+ if (!isAllowExplicitKeytab() && explicitKeytab != null) {
problems.add(new ValidationResult.Builder()
- .subject("Kerberos Credentials")
- .valid(false)
- .explanation("The '" + ALLOW_EXPLICIT_KEYTAB + "' system environment variable is configured to forbid explicitly configuring principal/keytab in processors. "
- + "The Kerberos Credentials Service should be used instead of setting the Kerberos Keytab or Kerberos Principal property.")
- .build());
+ .subject("Kerberos Credentials")
+ .valid(false)
+ .explanation("The '" + ALLOW_EXPLICIT_KEYTAB + "' system environment variable is configured to forbid explicitly configuring Kerberos Keytab in processors. "
+ + "The Kerberos Credentials Service should be used instead of setting the Kerberos Keytab or Kerberos Principal property.")
+ .build());
}
}
@@ -398,28 +405,37 @@ public class Hive3ConnectionPool extends AbstractControllerService implements Hi
if (SecurityUtil.isSecurityEnabled(hiveConfig)) {
final String explicitPrincipal = context.getProperty(kerberosProperties.getKerberosPrincipal()).evaluateAttributeExpressions().getValue();
final String explicitKeytab = context.getProperty(kerberosProperties.getKerberosKeytab()).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = context.getProperty(kerberosProperties.getKerberosPassword()).getValue();
final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
final String resolvedKeytab;
- if (credentialsService == null) {
- resolvedPrincipal = explicitPrincipal;
- resolvedKeytab = explicitKeytab;
- } else {
+ if (credentialsService != null) {
resolvedPrincipal = credentialsService.getPrincipal();
resolvedKeytab = credentialsService.getKeytab();
+ } else {
+ resolvedPrincipal = explicitPrincipal;
+ resolvedKeytab = explicitKeytab;
}
- log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ if (resolvedKeytab != null) {
+ kerberosUserReference.set(new KerberosKeytabUser(resolvedPrincipal, resolvedKeytab));
+ log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ } else if (explicitPassword != null) {
+ kerberosUserReference.set(new KerberosPasswordUser(resolvedPrincipal, explicitPassword));
+ log.info("Hive Security Enabled, logging in as principal {} with password", new Object[] {resolvedPrincipal});
+ } else {
+ throw new InitializationException("Unable to authenticate with Kerberos, no keytab or password was provided");
+ }
try {
- ugi = hiveConfigurator.authenticate(hiveConfig, resolvedPrincipal, resolvedKeytab);
+ ugi = hiveConfigurator.authenticate(hiveConfig, kerberosUserReference.get());
} catch (AuthenticationFailedException ae) {
log.error(ae.getMessage(), ae);
throw new InitializationException(ae);
}
- getLogger().info("Successfully logged in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ getLogger().info("Successfully logged in as principal " + resolvedPrincipal);
}
final String user = context.getProperty(DB_USER).evaluateAttributeExpressions().getValue();
@@ -476,13 +492,24 @@ public class Hive3ConnectionPool extends AbstractControllerService implements Hi
public Connection getConnection() throws ProcessException {
try {
if (ugi != null) {
- synchronized(this) {
- /*
- * Make sure that only one thread can request that the UGI relogin at a time. This
- * explicit relogin attempt is necessary due to the Hive client/thrift not implicitly handling
- * the acquisition of a new TGT after the current one has expired.
- * https://issues.apache.org/jira/browse/NIFI-5134
- */
+ /*
+ * Explicitly check the TGT and relogin if necessary with the KerberosUser instance. No synchronization
+ * is necessary in the client code, since AbstractKerberosUser's checkTGTAndRelogin method is synchronized.
+ */
+ getLogger().trace("getting UGI instance");
+ if (kerberosUserReference.get() != null) {
+ // if there's a KerberosUser associated with this UGI, check the TGT and relogin if it is close to expiring
+ KerberosUser kerberosUser = kerberosUserReference.get();
+ getLogger().debug("kerberosUser is " + kerberosUser);
+ try {
+ getLogger().debug("checking TGT on kerberosUser " + kerberosUser);
+ kerberosUser.checkTGTAndRelogin();
+ } catch (LoginException e) {
+ throw new ProcessException("Unable to relogin with kerberos credentials for " + kerberosUser.getPrincipal(), e);
+ }
+ } else {
+ getLogger().debug("kerberosUser was null, will not refresh TGT with KerberosUser");
+ // no synchronization is needed for UserGroupInformation.checkTGTAndReloginFromKeytab; UGI handles the synchronization internally
ugi.checkTGTAndReloginFromKeytab();
}
try {
@@ -515,4 +542,10 @@ public class Hive3ConnectionPool extends AbstractControllerService implements Hi
return connectionUrl;
}
+ /*
+ * Overridable by subclasses in the same package, mainly intended for testing purposes to allow verification without having to set environment variables.
+ */
+ boolean isAllowExplicitKeytab() {
+ return Boolean.parseBoolean(System.getenv(ALLOW_EXPLICIT_KEYTAB));
+ }
}
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java
index 535754f..a1123d2 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java
@@ -39,6 +39,7 @@ import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.expression.AttributeExpression;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.hadoop.SecurityUtil;
@@ -54,6 +55,9 @@ import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.processor.util.pattern.DiscontinuedException;
import org.apache.nifi.processor.util.pattern.RollbackOnFailure;
import org.apache.nifi.processors.hadoop.exception.RecordReaderFactoryException;
+import org.apache.nifi.security.krb.KerberosKeytabUser;
+import org.apache.nifi.security.krb.KerberosPasswordUser;
+import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.util.StringUtils;
@@ -201,6 +205,25 @@ public class PutHive3Streaming extends AbstractProcessor {
.required(false)
.build();
+ static final PropertyDescriptor KERBEROS_PRINCIPAL = new PropertyDescriptor.Builder()
+ .name("kerberos-principal")
+ .displayName("Kerberos Principal")
+ .description("The principal to use when specifying the principal and password directly in the processor for authenticating via Kerberos.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING))
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
+
+ static final PropertyDescriptor KERBEROS_PASSWORD = new PropertyDescriptor.Builder()
+ .name("kerberos-password")
+ .displayName("Kerberos Password")
+ .description("The password to use when specifying the principal and password directly in the processor for authenticating via Kerberos.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .sensitive(true)
+ .build();
+
// Relationships
public static final Relationship REL_SUCCESS = new Relationship.Builder()
.name("success")
@@ -225,6 +248,7 @@ public class PutHive3Streaming extends AbstractProcessor {
protected volatile HiveConfigurator hiveConfigurator = new HiveConfigurator();
protected volatile UserGroupInformation ugi;
+ final protected AtomicReference<KerberosUser> kerberosUserReference = new AtomicReference<>();
protected volatile HiveConf hiveConfig;
protected volatile int callTimeout;
@@ -247,6 +271,8 @@ public class PutHive3Streaming extends AbstractProcessor {
props.add(DISABLE_STREAMING_OPTIMIZATIONS);
props.add(ROLLBACK_ON_FAILURE);
props.add(KERBEROS_CREDENTIALS_SERVICE);
+ props.add(KERBEROS_PRINCIPAL);
+ props.add(KERBEROS_PASSWORD);
propertyDescriptors = Collections.unmodifiableList(props);
@@ -274,14 +300,23 @@ public class PutHive3Streaming extends AbstractProcessor {
final List<ValidationResult> problems = new ArrayList<>();
final KerberosCredentialsService credentialsService = validationContext.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
+ final String explicitPrincipal = validationContext.getProperty(KERBEROS_PRINCIPAL).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = validationContext.getProperty(KERBEROS_PASSWORD).getValue();
- final String resolvedPrincipal = credentialsService != null ? credentialsService.getPrincipal() : null;
+ final String resolvedPrincipal = credentialsService != null ? credentialsService.getPrincipal() : explicitPrincipal;
final String resolvedKeytab = credentialsService != null ? credentialsService.getKeytab() : null;
if (confFileProvided) {
final String configFiles = validationContext.getProperty(HIVE_CONFIGURATION_RESOURCES).evaluateAttributeExpressions().getValue();
- problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, validationResourceHolder, getLogger()));
+ problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, explicitPassword, validationResourceHolder, getLogger()));
}
+ if (credentialsService != null && (explicitPrincipal != null || explicitPassword != null)) {
+ problems.add(new ValidationResult.Builder()
+ .subject(KERBEROS_CREDENTIALS_SERVICE.getDisplayName())
+ .valid(false)
+ .explanation("kerberos principal/password and kerberos credential service cannot be configured at the same time")
+ .build());
+ }
return problems;
}
@@ -310,20 +345,33 @@ public class PutHive3Streaming extends AbstractProcessor {
if (SecurityUtil.isSecurityEnabled(hiveConfig)) {
final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
+ final String explicitPrincipal = context.getProperty(KERBEROS_PRINCIPAL).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = context.getProperty(KERBEROS_PASSWORD).getValue();
+
+ final String resolvedPrincipal = credentialsService != null ? credentialsService.getPrincipal() : explicitPrincipal;
+ final String resolvedKeytab = credentialsService != null ? credentialsService.getKeytab() : null;
+
+ if (resolvedKeytab != null) {
+ kerberosUserReference.set(new KerberosKeytabUser(resolvedPrincipal, resolvedKeytab));
+ log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ } else if (explicitPassword != null) {
+ kerberosUserReference.set(new KerberosPasswordUser(resolvedPrincipal, explicitPassword));
+ log.info("Hive Security Enabled, logging in as principal {} with password", new Object[] {resolvedPrincipal});
+ } else {
+ throw new ProcessException("Unable to authenticate with Kerberos, no keytab or password was provided");
+ }
- final String resolvedPrincipal = credentialsService.getPrincipal();
- final String resolvedKeytab = credentialsService.getKeytab();
-
- log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[]{resolvedPrincipal, resolvedKeytab});
try {
- ugi = hiveConfigurator.authenticate(hiveConfig, resolvedPrincipal, resolvedKeytab);
+ ugi = hiveConfigurator.authenticate(hiveConfig, kerberosUserReference.get());
} catch (AuthenticationFailedException ae) {
- throw new ProcessException("Kerberos authentication failed for Hive Streaming", ae);
+ log.error(ae.getMessage(), ae);
+ throw new ProcessException(ae);
}
log.info("Successfully logged in as principal {} with keytab {}", new Object[]{resolvedPrincipal, resolvedKeytab});
} else {
ugi = null;
+ kerberosUserReference.set(null);
}
callTimeout = context.getProperty(CALL_TIMEOUT).evaluateAttributeExpressions().asInteger() * 1000; // milliseconds
@@ -532,6 +580,7 @@ public class PutHive3Streaming extends AbstractProcessor {
}
ugi = null;
+ kerberosUserReference.set(null);
}
private void abortAndCloseConnection(StreamingConnection connection) {
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
index 092557b..3832944 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
@@ -26,6 +26,7 @@ import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.hadoop.KerberosProperties;
import org.apache.nifi.hadoop.SecurityUtil;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.security.krb.KerberosUser;
import java.io.IOException;
import java.util.ArrayList;
@@ -38,7 +39,8 @@ import java.util.concurrent.atomic.AtomicReference;
*/
public class HiveConfigurator {
- public Collection<ValidationResult> validate(String configFiles, String principal, String keyTab, AtomicReference<ValidationResources> validationResourceHolder, ComponentLog log) {
+ public Collection<ValidationResult> validate(String configFiles, String principal, String keyTab, String password,
+ AtomicReference<ValidationResources> validationResourceHolder, ComponentLog log) {
final List<ValidationResult> problems = new ArrayList<>();
ValidationResources resources = validationResourceHolder.get();
@@ -53,7 +55,7 @@ public class HiveConfigurator {
final Configuration hiveConfig = resources.getConfiguration();
- problems.addAll(KerberosProperties.validatePrincipalWithKeytabOrPassword(this.getClass().getSimpleName(), hiveConfig, principal, keyTab, null, log));
+ problems.addAll(KerberosProperties.validatePrincipalWithKeytabOrPassword(this.getClass().getSimpleName(), hiveConfig, principal, keyTab, password, log));
return problems;
}
@@ -78,6 +80,22 @@ public class HiveConfigurator {
}
/**
+ * Acquires a {@link UserGroupInformation} using the given {@link Configuration} and {@link KerberosUser}.
+ * @see SecurityUtil#getUgiForKerberosUser(Configuration, KerberosUser)
+ * @param hiveConfig The Configuration to apply to the acquired UserGroupInformation
+ * @param kerberosUser The KerberosUser to authenticate
+ * @return A UserGroupInformation instance created using the Subject of the given KerberosUser
+ * @throws AuthenticationFailedException if authentication fails
+ */
+ public UserGroupInformation authenticate(final Configuration hiveConfig, KerberosUser kerberosUser) throws AuthenticationFailedException {
+ try {
+ return SecurityUtil.getUgiForKerberosUser(hiveConfig, kerberosUser);
+ } catch (IOException ioe) {
+ throw new AuthenticationFailedException("Kerberos Authentication for Hive failed", ioe);
+ }
+ }
+
+ /**
* As of Apache NiFi 1.5.0, due to changes made to
* {@link SecurityUtil#loginKerberos(Configuration, String, String)}, which is used by this
* class to authenticate a principal with Kerberos, Hive controller services no longer
@@ -94,7 +112,9 @@ public class HiveConfigurator {
* authentication attempts that would leave the Hive controller service in an unrecoverable state.
*
* @see SecurityUtil#loginKerberos(Configuration, String, String)
+ * @deprecated Use {@link SecurityUtil#getUgiForKerberosUser(Configuration, KerberosUser)}
*/
+ @Deprecated
public UserGroupInformation authenticate(final Configuration hiveConfig, String principal, String keyTab) throws AuthenticationFailedException {
UserGroupInformation ugi;
try {
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java
index 16c42e5..d9113fe 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java
@@ -61,6 +61,7 @@ import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.schema.access.SchemaAccessUtils;
import org.apache.nifi.schema.access.SchemaNotFoundException;
+import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.record.MapRecord;
@@ -115,9 +116,12 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
@@ -262,7 +266,7 @@ public class TestPutHive3Streaming {
}
@Test
- public void testUgiGetsCleared() throws Exception {
+ public void testUgiAndKerberosUserGetsCleared() throws Exception {
configure(processor, 0);
runner.setProperty(PutHive3Streaming.METASTORE_URI, "thrift://localhost:9083");
runner.setProperty(PutHive3Streaming.DB_NAME, "default");
@@ -270,6 +274,7 @@ public class TestPutHive3Streaming {
processor.ugi = mock(UserGroupInformation.class);
runner.run();
assertNull(processor.ugi);
+ assertNull(processor.kerberosUserReference.get());
}
@Test
@@ -281,12 +286,13 @@ public class TestPutHive3Streaming {
runner.setProperty(KERBEROS_CREDENTIALS_SERVICE, "kcs");
runner.enableControllerService(kcs);
ugi = mock(UserGroupInformation.class);
- when(hiveConfigurator.authenticate(eq(hiveConf), anyString(), anyString())).thenReturn(ugi);
+ when(hiveConfigurator.authenticate(eq(hiveConf), any(KerberosUser.class))).thenReturn(ugi);
runner.setProperty(PutHive3Streaming.METASTORE_URI, "thrift://localhost:9083");
runner.setProperty(PutHive3Streaming.DB_NAME, "default");
runner.setProperty(PutHive3Streaming.TABLE_NAME, "users");
runner.enqueue(new byte[0]);
runner.run();
+ verify(hiveConfigurator, times(1)).authenticate(eq(hiveConf), any(KerberosUser.class));
}
@Test(expected = AssertionError.class)
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive_1_1ConnectionPool.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive_1_1ConnectionPool.java
index dd2e1fe..1e460bc 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive_1_1ConnectionPool.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/dbcp/hive/Hive_1_1ConnectionPool.java
@@ -31,12 +31,16 @@ import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.expression.AttributeExpression;
import org.apache.nifi.hadoop.SecurityUtil;
import org.apache.nifi.kerberos.KerberosCredentialsService;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.security.krb.KerberosKeytabUser;
+import org.apache.nifi.security.krb.KerberosPasswordUser;
+import org.apache.nifi.security.krb.KerberosUser;
import org.apache.nifi.util.hive.AuthenticationFailedException;
import org.apache.nifi.util.hive.HiveConfigurator;
import org.apache.nifi.util.hive.HiveUtils;
@@ -57,6 +61,8 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.nifi.controller.ControllerServiceInitializationContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
+import javax.security.auth.login.LoginException;
+
/**
* Implementation for Database Connection Pooling Service used for Apache Hive 1.1
* connections. Apache DBCP is used for connection pooling functionality.
@@ -149,6 +155,25 @@ public class Hive_1_1ConnectionPool extends AbstractControllerService implements
.required(false)
.build();
+ static final PropertyDescriptor KERBEROS_PRINCIPAL = new PropertyDescriptor.Builder()
+ .name("kerberos-principal")
+ .displayName("Kerberos Principal")
+ .description("The principal to use when specifying the principal and password directly in the processor for authenticating via Kerberos.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(AttributeExpression.ResultType.STRING))
+ .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
+
+ static final PropertyDescriptor KERBEROS_PASSWORD = new PropertyDescriptor.Builder()
+ .name("kerberos-password")
+ .displayName("Kerberos Password")
+ .description("The password to use when specifying the principal and password directly in the processor for authenticating via Kerberos.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .sensitive(true)
+ .build();
+
private List<PropertyDescriptor> properties;
@@ -161,6 +186,7 @@ public class Hive_1_1ConnectionPool extends AbstractControllerService implements
private volatile HiveConfigurator hiveConfigurator = new HiveConfigurator();
private volatile UserGroupInformation ugi;
+ private final AtomicReference<KerberosUser> kerberosUserReference = new AtomicReference<>();
@Override
protected void init(final ControllerServiceInitializationContext context) {
@@ -173,6 +199,8 @@ public class Hive_1_1ConnectionPool extends AbstractControllerService implements
props.add(MAX_TOTAL_CONNECTIONS);
props.add(VALIDATION_QUERY);
props.add(KERBEROS_CREDENTIALS_SERVICE);
+ props.add(KERBEROS_PRINCIPAL);
+ props.add(KERBEROS_PASSWORD);
properties = props;
}
@@ -190,19 +218,29 @@ public class Hive_1_1ConnectionPool extends AbstractControllerService implements
if (confFileProvided) {
final KerberosCredentialsService credentialsService = validationContext.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
+ final String explicitPrincipal = validationContext.getProperty(KERBEROS_PRINCIPAL).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = validationContext.getProperty(KERBEROS_PASSWORD).getValue();
final String resolvedPrincipal;
final String resolvedKeytab;
- if (credentialsService == null) {
- resolvedPrincipal = null;
- resolvedKeytab = null;
- } else {
+ if (credentialsService != null) {
resolvedPrincipal = credentialsService.getPrincipal();
resolvedKeytab = credentialsService.getKeytab();
+ } else {
+ resolvedPrincipal = explicitPrincipal;
+ resolvedKeytab = null;
}
final String configFiles = validationContext.getProperty(HIVE_CONFIGURATION_RESOURCES).evaluateAttributeExpressions().getValue();
- problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, validationResourceHolder, getLogger()));
+ problems.addAll(hiveConfigurator.validate(configFiles, resolvedPrincipal, resolvedKeytab, explicitPassword, validationResourceHolder, getLogger()));
+
+ if (credentialsService != null && (explicitPrincipal != null || explicitPassword != null)) {
+ problems.add(new ValidationResult.Builder()
+ .subject(KERBEROS_CREDENTIALS_SERVICE.getDisplayName())
+ .valid(false)
+ .explanation("kerberos principal/password and kerberos credential service cannot be configured at the same time")
+ .build());
+ }
}
return problems;
@@ -260,28 +298,38 @@ public class Hive_1_1ConnectionPool extends AbstractControllerService implements
final String drv = HiveDriver.class.getName();
if (SecurityUtil.isSecurityEnabled(hiveConfig)) {
+ final String explicitPrincipal = context.getProperty(KERBEROS_PRINCIPAL).evaluateAttributeExpressions().getValue();
+ final String explicitPassword = context.getProperty(KERBEROS_PASSWORD).getValue();
final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
final String resolvedPrincipal;
final String resolvedKeytab;
- if (credentialsService == null) {
- resolvedPrincipal = null;
- resolvedKeytab = null;
- } else {
+ if (credentialsService != null) {
resolvedPrincipal = credentialsService.getPrincipal();
resolvedKeytab = credentialsService.getKeytab();
+ } else {
+ resolvedPrincipal = explicitPrincipal;
+ resolvedKeytab = null;
}
- log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ if (resolvedKeytab != null) {
+ kerberosUserReference.set(new KerberosKeytabUser(resolvedPrincipal, resolvedKeytab));
+ log.info("Hive Security Enabled, logging in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ } else if (explicitPassword != null) {
+ kerberosUserReference.set(new KerberosPasswordUser(resolvedPrincipal, explicitPassword));
+ log.info("Hive Security Enabled, logging in as principal {} with password", new Object[] {resolvedPrincipal});
+ } else {
+ throw new InitializationException("Unable to authenticate with Kerberos, no keytab or password was provided");
+ }
try {
- ugi = hiveConfigurator.authenticate(hiveConfig, resolvedPrincipal, resolvedKeytab);
+ ugi = hiveConfigurator.authenticate(hiveConfig, kerberosUserReference.get());
} catch (AuthenticationFailedException ae) {
log.error(ae.getMessage(), ae);
throw new InitializationException(ae);
}
- getLogger().info("Successfully logged in as principal {} with keytab {}", new Object[] {resolvedPrincipal, resolvedKeytab});
+ getLogger().info("Successfully logged in as principal " + resolvedPrincipal);
}
final String user = context.getProperty(DB_USER).evaluateAttributeExpressions().getValue();
@@ -325,13 +373,24 @@ public class Hive_1_1ConnectionPool extends AbstractControllerService implements
public Connection getConnection() throws ProcessException {
try {
if (ugi != null) {
- synchronized(this) {
- /*
- * Make sure that only one thread can request that the UGI relogin at a time. This
- * explicit relogin attempt is necessary due to the Hive client/thrift not implicitly handling
- * the acquisition of a new TGT after the current one has expired.
- * https://issues.apache.org/jira/browse/NIFI-5134
- */
+ /*
+ * Explicitly check the TGT and relogin if necessary with the KerberosUser instance. No synchronization
+ * is necessary in the client code, since AbstractKerberosUser's checkTGTAndRelogin method is synchronized.
+ */
+ getLogger().trace("getting UGI instance");
+ if (kerberosUserReference.get() != null) {
+ // if there's a KerberosUser associated with this UGI, check the TGT and relogin if it is close to expiring
+ KerberosUser kerberosUser = kerberosUserReference.get();
+ getLogger().debug("kerberosUser is " + kerberosUser);
+ try {
+ getLogger().debug("checking TGT on kerberosUser " + kerberosUser);
+ kerberosUser.checkTGTAndRelogin();
+ } catch (LoginException e) {
+ throw new ProcessException("Unable to relogin with kerberos credentials for " + kerberosUser.getPrincipal(), e);
+ }
+ } else {
+ getLogger().debug("kerberosUser was null, will not refresh TGT with KerberosUser");
+ // no synchronization is needed for UserGroupInformation.checkTGTAndReloginFromKeytab; UGI handles the synchronization internally
ugi.checkTGTAndReloginFromKeytab();
}
try {
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
index 5fed81a..b212207 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive_1_1-processors/src/main/java/org/apache/nifi/util/hive/HiveConfigurator.java
@@ -26,6 +26,7 @@ import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.hadoop.KerberosProperties;
import org.apache.nifi.hadoop.SecurityUtil;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.security.krb.KerberosUser;
import java.io.IOException;
import java.util.ArrayList;
@@ -35,7 +36,8 @@ import java.util.concurrent.atomic.AtomicReference;
public class HiveConfigurator {
- public Collection<ValidationResult> validate(String configFiles, String principal, String keyTab, AtomicReference<ValidationResources> validationResourceHolder, ComponentLog log) {
+ public Collection<ValidationResult> validate(String configFiles, String principal, String keyTab, String password,
+ AtomicReference<ValidationResources> validationResourceHolder, ComponentLog log) {
final List<ValidationResult> problems = new ArrayList<>();
ValidationResources resources = validationResourceHolder.get();
@@ -50,7 +52,7 @@ public class HiveConfigurator {
final Configuration hiveConfig = resources.getConfiguration();
- problems.addAll(KerberosProperties.validatePrincipalWithKeytabOrPassword(this.getClass().getSimpleName(), hiveConfig, principal, keyTab, null, log));
+ problems.addAll(KerberosProperties.validatePrincipalWithKeytabOrPassword(this.getClass().getSimpleName(), hiveConfig, principal, keyTab, password, log));
return problems;
}
@@ -75,6 +77,22 @@ public class HiveConfigurator {
}
/**
+ * Acquires a {@link UserGroupInformation} using the given {@link Configuration} and {@link KerberosUser}.
+ * @see SecurityUtil#getUgiForKerberosUser(Configuration, KerberosUser)
+ * @param hiveConfig The Configuration to apply to the acquired UserGroupInformation
+ * @param kerberosUser The KerberosUser to authenticate
+ * @return A UserGroupInformation instance created using the Subject of the given KerberosUser
+ * @throws AuthenticationFailedException if authentication fails
+ */
+ public UserGroupInformation authenticate(final Configuration hiveConfig, KerberosUser kerberosUser) throws AuthenticationFailedException {
+ try {
+ return SecurityUtil.getUgiForKerberosUser(hiveConfig, kerberosUser);
+ } catch (IOException ioe) {
+ throw new AuthenticationFailedException("Kerberos Authentication for Hive failed", ioe);
+ }
+ }
+
+ /**
* As of Apache NiFi 1.5.0, due to changes made to
* {@link SecurityUtil#loginKerberos(Configuration, String, String)}, which is used by this
* class to authenticate a principal with Kerberos, Hive controller services no longer
@@ -91,7 +109,9 @@ public class HiveConfigurator {
* authentication attempts that would leave the Hive controller service in an unrecoverable state.
*
* @see SecurityUtil#loginKerberos(Configuration, String, String)
+ * @deprecated Use {@link SecurityUtil#getUgiForKerberosUser(Configuration, KerberosUser)}
*/
+ @Deprecated
public UserGroupInformation authenticate(final Configuration hiveConfig, String principal, String keyTab) throws AuthenticationFailedException {
UserGroupInformation ugi;
try {