KAFKA-10895: Gracefully handle invalid JAAS configs (follow up fix) (#9987)

Fixes a regression introduced by https://github.com/apache/kafka/pull/9806 in the original fix for KAFKA-10895

It was discovered that if an invalid JAAS config is present on the worker, invoking Configuration::getConfiguration throws an exception. The changes from #9806 cause that exception to be thrown during plugin scanning, which causes the worker to fail even if it is not configured to use the basic auth extension at all.

This follow-up handles invalid JAAS configurations more gracefully, and only throws them if the worker is actually configured to use the basic auth extension, at the time that the extension is instantiated and configured.

Two unit tests are added.

Reviewers: Greg Harris <gregh@confluent.io>, Konstantine Karantasis <k.karantasis@gmail.com>
This commit is contained in:
Chris Egerton 2021-02-03 16:51:03 -05:00 committed by GitHub
parent 772f2cfc82
commit 0fe9fde376
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 79 additions and 3 deletions

View File

@ -18,6 +18,7 @@
package org.apache.kafka.connect.rest.basic.auth.extension; package org.apache.kafka.connect.rest.basic.auth.extension;
import org.apache.kafka.common.utils.AppInfoParser; import org.apache.kafka.common.utils.AppInfoParser;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.rest.ConnectRestExtension; import org.apache.kafka.connect.rest.ConnectRestExtension;
import org.apache.kafka.connect.rest.ConnectRestExtensionContext; import org.apache.kafka.connect.rest.ConnectRestExtensionContext;
import org.slf4j.Logger; import org.slf4j.Logger;
@ -26,6 +27,7 @@ import org.slf4j.LoggerFactory;
import javax.security.auth.login.Configuration; import javax.security.auth.login.Configuration;
import java.io.IOException; import java.io.IOException;
import java.util.Map; import java.util.Map;
import java.util.function.Supplier;
/** /**
* Provides the ability to authenticate incoming BasicAuth credentials using the configured JAAS {@link * Provides the ability to authenticate incoming BasicAuth credentials using the configured JAAS {@link
@ -62,14 +64,38 @@ public class BasicAuthSecurityRestExtension implements ConnectRestExtension {
private static final Logger log = LoggerFactory.getLogger(BasicAuthSecurityRestExtension.class); private static final Logger log = LoggerFactory.getLogger(BasicAuthSecurityRestExtension.class);
private static final Supplier<Configuration> CONFIGURATION = initializeConfiguration(Configuration::getConfiguration);
// Capture the JVM's global JAAS configuration as soon as possible, as it may be altered later // Capture the JVM's global JAAS configuration as soon as possible, as it may be altered later
// by connectors, converters, other REST extensions, etc. // by connectors, converters, other REST extensions, etc.
private static final Configuration CONFIGURATION = Configuration.getConfiguration(); static Supplier<Configuration> initializeConfiguration(Supplier<Configuration> configurationSupplier) {
try {
Configuration configuration = configurationSupplier.get();
return () -> configuration;
} catch (Exception e) {
// We have to be careful not to throw anything here as this static block gets executed during plugin scanning and any exceptions will
// cause the worker to fail during startup, even if it's not configured to use the basic auth extension.
return () -> {
throw new ConnectException("Failed to retrieve JAAS configuration", e);
};
}
}
private final Supplier<Configuration> configuration;
public BasicAuthSecurityRestExtension() {
this(CONFIGURATION);
}
// For testing
BasicAuthSecurityRestExtension(Supplier<Configuration> configuration) {
this.configuration = configuration;
}
@Override @Override
public void register(ConnectRestExtensionContext restPluginContext) { public void register(ConnectRestExtensionContext restPluginContext) {
log.trace("Registering JAAS basic auth filter"); log.trace("Registering JAAS basic auth filter");
restPluginContext.configurable().register(new JaasBasicAuthFilter(CONFIGURATION)); restPluginContext.configurable().register(new JaasBasicAuthFilter(configuration.get()));
log.trace("Finished registering JAAS basic auth filter"); log.trace("Finished registering JAAS basic auth filter");
} }
@ -80,7 +106,8 @@ public class BasicAuthSecurityRestExtension implements ConnectRestExtension {
@Override @Override
public void configure(Map<String, ?> configs) { public void configure(Map<String, ?> configs) {
// If we failed to retrieve a JAAS configuration during startup, throw that exception now
configuration.get();
} }
@Override @Override

View File

@ -17,6 +17,7 @@
package org.apache.kafka.connect.rest.basic.auth.extension; package org.apache.kafka.connect.rest.basic.auth.extension;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.rest.ConnectRestExtensionContext; import org.apache.kafka.connect.rest.ConnectRestExtensionContext;
import org.easymock.Capture; import org.easymock.Capture;
import org.easymock.EasyMock; import org.easymock.EasyMock;
@ -27,7 +28,15 @@ import org.junit.jupiter.api.Test;
import javax.security.auth.login.Configuration; import javax.security.auth.login.Configuration;
import javax.ws.rs.core.Configurable; import javax.ws.rs.core.Configurable;
import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
public class BasicAuthSecurityRestExtensionTest { public class BasicAuthSecurityRestExtensionTest {
@ -63,4 +72,44 @@ public class BasicAuthSecurityRestExtensionTest {
assertNotEquals(overwrittenConfiguration, jaasFilter.getValue().configuration, assertNotEquals(overwrittenConfiguration, jaasFilter.getValue().configuration,
"Overwritten JAAS configuration should not be used by basic auth REST extension"); "Overwritten JAAS configuration should not be used by basic auth REST extension");
} }
@Test
public void testBadJaasConfigInitialization() {
SecurityException jaasConfigurationException = new SecurityException(new IOException("Bad JAAS config is bad"));
Supplier<Configuration> configuration = BasicAuthSecurityRestExtension.initializeConfiguration(() -> {
throw jaasConfigurationException;
});
ConnectException thrownException = assertThrows(ConnectException.class, configuration::get);
assertEquals(jaasConfigurationException, thrownException.getCause());
}
@Test
public void testGoodJaasConfigInitialization() {
AtomicBoolean configurationInitializerEvaluated = new AtomicBoolean(false);
Configuration mockConfiguration = EasyMock.mock(Configuration.class);
Supplier<Configuration> configuration = BasicAuthSecurityRestExtension.initializeConfiguration(() -> {
configurationInitializerEvaluated.set(true);
return mockConfiguration;
});
assertTrue(configurationInitializerEvaluated.get());
assertEquals(mockConfiguration, configuration.get());
}
@Test
public void testBadJaasConfigExtensionSetup() {
SecurityException jaasConfigurationException = new SecurityException(new IOException("Bad JAAS config is bad"));
Supplier<Configuration> configuration = () -> {
throw jaasConfigurationException;
};
BasicAuthSecurityRestExtension extension = new BasicAuthSecurityRestExtension(configuration);
Exception thrownException = assertThrows(Exception.class, () -> extension.configure(Collections.emptyMap()));
assertEquals(jaasConfigurationException, thrownException);
thrownException = assertThrows(Exception.class, () -> extension.register(EasyMock.mock(ConnectRestExtensionContext.class)));
assertEquals(jaasConfigurationException, thrownException);
}
} }