SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Kafka binder JAAS config refactoring (#461)

Browse Source 
* Kafka binder JAAS config refactoring

Refactor JaasLoginModuleConfiguraton to leverage ControlFlag enum from
spring-kafka, instead of directly using LoginModuleControlFlag from JDK.

Re-instate Jass configuration tests.

Resolves #459

* Fixing tests

* Addressing PR review comments
This commit is contained in:
Soby Chacko
2018-10-10 14:53:13 -04:00
committed by Gary Russell
parent fe7bdd245d
commit 9443c621b0
5 changed files with 149 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pom.xml
View File

@@ -12,7 +12,7 @@
</parent>
<properties>
<java.version>1.8</java.version>
<spring-kafka.version>2.2.0.RC1</spring-kafka.version>
<spring-kafka.version>2.2.0.BUILD-SNAPSHOT</spring-kafka.version>
<spring-integration-kafka.version>3.1.0.M1</spring-integration-kafka.version>
<kafka.version>2.0.0</kafka.version>
<spring-cloud-stream.version>2.1.0.BUILD-SNAPSHOT</spring-cloud-stream.version>

28
spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import java.util.Map;
import javax.security.auth.login.AppConfigurationEntry;
import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
import org.springframework.util.Assert;
/**
@@ -28,12 +29,13 @@ import org.springframework.util.Assert;
* for the Kafka or Zookeeper client.
*
* @author Marius Bogoevici
* @author Soby Chacko
*/
public class JaasLoginModuleConfiguration {
private String loginModule = "com.sun.security.auth.module.Krb5LoginModule";
private AppConfigurationEntry.LoginModuleControlFlag controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUIRED;
private KafkaJaasLoginModuleInitializer.ControlFlag controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.REQUIRED;
private Map<String,String> options = new HashMap<>();
@@ -46,31 +48,13 @@ public class JaasLoginModuleConfiguration {
this.loginModule = loginModule;
}
public String getControlFlag() {
return controlFlag.toString();
}
public AppConfigurationEntry.LoginModuleControlFlag getControlFlagValue() {
public KafkaJaasLoginModuleInitializer.ControlFlag getControlFlag() {
return controlFlag;
}
public void setControlFlag(String controlFlag) {
Assert.notNull(controlFlag, "cannot be null");
if (AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL.equals(controlFlag)) {
this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL;
}
else if (AppConfigurationEntry.LoginModuleControlFlag.REQUIRED.equals(controlFlag)) {
this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUIRED;
}
else if (AppConfigurationEntry.LoginModuleControlFlag.REQUISITE.equals(controlFlag)) {
this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUISITE;
}
else if (AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT.equals(controlFlag)) {
this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT;
}
else {
throw new IllegalArgumentException(controlFlag + " is not a supported control flag");
}
this.controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.valueOf(controlFlag.toUpperCase());
}
public Map<String, String> getOptions() {

17
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
View File

@@ -18,8 +18,6 @@ package org.springframework.cloud.stream.binder.kafka.config;
import java.io.IOException;
import javax.security.auth.login.AppConfigurationEntry;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.binder.MeterBinder;
@@ -113,21 +111,8 @@ public class KafkaBinderConfiguration {
if (jaas != null) {
kafkaJaasLoginModuleInitializer.setLoginModule(jaas.getLoginModule());
KafkaJaasLoginModuleInitializer.ControlFlag controlFlag = null;
AppConfigurationEntry.LoginModuleControlFlag controlFlagValue = jaas.getControlFlagValue();
KafkaJaasLoginModuleInitializer.ControlFlag controlFlag = jaas.getControlFlag();
if (AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL.equals(controlFlagValue)) {
controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.OPTIONAL;
}
else if (AppConfigurationEntry.LoginModuleControlFlag.REQUIRED.equals(controlFlagValue)) {
controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.REQUIRED;
}
else if (AppConfigurationEntry.LoginModuleControlFlag.REQUISITE.equals(controlFlagValue)) {
controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.REQUISITE;
}
else if (AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT.equals(controlFlagValue)) {
controlFlag = KafkaJaasLoginModuleInitializer.ControlFlag.SUFFICIENT;
}
if (controlFlag != null) {
kafkaJaasLoginModuleInitializer.setControlFlag(controlFlag);
}

122
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java Normal file
View File

@@ -0,0 +1,122 @@
/*
* Copyright 2016-2018 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binder.kafka;
import javax.security.auth.login.AppConfigurationEntry;
import com.sun.security.auth.login.ConfigFile;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.ConfigurationPropertiesBindException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* @author Marius Bogoevici
* @author Soby Chacko
*/
public class KafkaBinderJaasInitializerListenerTest {
private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";
@ClassRule
public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true);
@BeforeClass
public static void setup() {
System.setProperty(KAFKA_BROKERS_PROPERTY, kafkaEmbedded.getEmbeddedKafka().getBrokersAsString());
}
@AfterClass
public static void clean() {
System.clearProperty(KAFKA_BROKERS_PROPERTY);
}
@Test
public void testConfigurationParsedCorrectlyWithKafkaClientAndDefaultControlFlag() throws Exception {
ConfigFile configFile = new ConfigFile(new ClassPathResource("jaas-sample-kafka-only.conf").getURI());
final AppConfigurationEntry[] kafkaConfigurationArray = configFile.getAppConfigurationEntry("KafkaClient");
final ConfigurableApplicationContext context =
SpringApplication.run(SimpleApplication.class,
"--spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true",
"--spring.cloud.stream.kafka.binder.jaas.options.storeKey=true",
"--spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab",
"--spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM",
"--spring.jmx.enabled=false");
javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration.getConfiguration();
final AppConfigurationEntry[] kafkaConfiguration = configuration.getAppConfigurationEntry("KafkaClient");
assertThat(kafkaConfiguration).hasSize(1);
assertThat(kafkaConfiguration[0].getOptions()).isEqualTo(kafkaConfigurationArray[0].getOptions());
assertThat(kafkaConfiguration[0].getControlFlag()).isEqualTo(AppConfigurationEntry.LoginModuleControlFlag.REQUIRED);
context.close();
}
@Test
public void testConfigurationParsedCorrectlyWithKafkaClientAndNonDefaultControlFlag() throws Exception {
ConfigFile configFile = new ConfigFile(new ClassPathResource("jaas-sample-kafka-only.conf").getURI());
final AppConfigurationEntry[] kafkaConfigurationArray = configFile.getAppConfigurationEntry("KafkaClient");
final ConfigurableApplicationContext context =
SpringApplication.run(SimpleApplication.class,
"--spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true",
"--spring.cloud.stream.kafka.binder.jaas.controlFlag=requisite",
"--spring.cloud.stream.kafka.binder.jaas.options.storeKey=true",
"--spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab",
"--spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM",
"--spring.jmx.enabled=false");
javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration.getConfiguration();
final AppConfigurationEntry[] kafkaConfiguration = configuration.getAppConfigurationEntry("KafkaClient");
assertThat(kafkaConfiguration).hasSize(1);
assertThat(kafkaConfiguration[0].getOptions()).isEqualTo(kafkaConfigurationArray[0].getOptions());
assertThat(kafkaConfiguration[0].getControlFlag()).isEqualTo(AppConfigurationEntry.LoginModuleControlFlag.REQUISITE);
context.close();
}
@Test
public void testConfigurationWithUnknownControlFlag() throws Exception {
ConfigFile configFile = new ConfigFile(new ClassPathResource("jaas-sample-kafka-only.conf").getURI());
assertThatThrownBy(
() -> SpringApplication.run(SimpleApplication.class,
"--spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true",
"--spring.cloud.stream.kafka.binder.jaas.controlFlag=unknown",
"--spring.cloud.stream.kafka.binder.jaas.options.storeKey=true",
"--spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab",
"--spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM",
"--spring.jmx.enabled=false"))
.isInstanceOf(ConfigurationPropertiesBindException.class)
.hasMessageContaining("Error creating bean with name 'configurationProperties'");
}
@SpringBootApplication
public static class SimpleApplication {
}
}

39
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java
View File

@@ -36,8 +36,6 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.messaging.MessageChannel;
@@ -52,20 +50,19 @@ import static org.assertj.core.api.Assertions.assertThat;
*/
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
properties = {"spring.cloud.stream.kafka.bindings.output.producer.configuration.key.serializer=FooSerializer.class",
properties = {"spring.cloud.stream.kafka.bindings.standard-out.producer.configuration.key.serializer=FooSerializer.class",
"spring.cloud.stream.kafka.default.producer.configuration.key.serializer=BarSerializer.class",
"spring.cloud.stream.kafka.default.producer.configuration.value.serializer=BarSerializer.class",
"spring.cloud.stream.kafka.bindings.input.consumer.configuration.key.serializer=FooSerializer.class",
"spring.cloud.stream.kafka.bindings.standard-in.consumer.configuration.key.serializer=FooSerializer.class",
"spring.cloud.stream.kafka.default.consumer.configuration.key.serializer=BarSerializer.class",
"spring.cloud.stream.kafka.default.consumer.configuration.value.serializer=BarSerializer.class",
"spring.cloud.stream.kafka.default.producer.configuration.foo=bar",
"spring.cloud.stream.kafka.bindings.output.producer.configuration.foo=bindingSpecificPropertyShouldWinOverDefault",
"spring.cloud.stream.kafka.bindings.standard-out.producer.configuration.foo=bindingSpecificPropertyShouldWinOverDefault",
"spring.cloud.stream.kafka.default.consumer.ackEachRecord=true",
"spring.cloud.stream.kafka.bindings.custom-in.consumer.ackEachRecord=false"})
public class KafkaBinderExtendedPropertiesTest {
private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";
private static final String ZK_NODE_PROPERTY = "spring.cloud.stream.kafka.binder.zkNodes";
@ClassRule
public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true);
@@ -73,13 +70,11 @@ public class KafkaBinderExtendedPropertiesTest {
@BeforeClass
public static void setup() {
System.setProperty(KAFKA_BROKERS_PROPERTY, kafkaEmbedded.getEmbeddedKafka().getBrokersAsString());
System.setProperty(ZK_NODE_PROPERTY, kafkaEmbedded.getEmbeddedKafka().getZookeeperConnectionString());
}
@AfterClass
public static void clean() {
System.clearProperty(KAFKA_BROKERS_PROPERTY);
System.clearProperty(ZK_NODE_PROPERTY);
}
@Autowired
@@ -93,25 +88,25 @@ public class KafkaBinderExtendedPropertiesTest {
binderFactory.getBinder("kafka", MessageChannel.class);
KafkaProducerProperties kafkaProducerProperties =
(KafkaProducerProperties)((ExtendedPropertiesBinder) kafkaBinder).getExtendedProducerProperties("output");
(KafkaProducerProperties)((ExtendedPropertiesBinder) kafkaBinder).getExtendedProducerProperties("standard-out");
//binding "output" gets FooSerializer defined on the binding itself and BarSerializer through default property.
//binding "standard-out" gets FooSerializer defined on the binding itself and BarSerializer through default property.
assertThat(kafkaProducerProperties.getConfiguration().get("key.serializer")).isEqualTo("FooSerializer.class");
assertThat(kafkaProducerProperties.getConfiguration().get("value.serializer")).isEqualTo("BarSerializer.class");
assertThat(kafkaProducerProperties.getConfiguration().get("foo")).isEqualTo("bindingSpecificPropertyShouldWinOverDefault");
KafkaConsumerProperties kafkaConsumerProperties =
(KafkaConsumerProperties)((ExtendedPropertiesBinder) kafkaBinder).getExtendedConsumerProperties("input");
(KafkaConsumerProperties)((ExtendedPropertiesBinder) kafkaBinder).getExtendedConsumerProperties("standard-in");
//binding "input" gets FooSerializer defined on the binding itself and BarSerializer through default property.
//binding "standard-in" gets FooSerializer defined on the binding itself and BarSerializer through default property.
assertThat(kafkaConsumerProperties.getConfiguration().get("key.serializer")).isEqualTo("FooSerializer.class");
assertThat(kafkaConsumerProperties.getConfiguration().get("value.serializer")).isEqualTo("BarSerializer.class");
KafkaProducerProperties customKafkaProducerProperties =
(KafkaProducerProperties)((ExtendedPropertiesBinder) kafkaBinder).getExtendedProducerProperties("custom-out");
//binding "output" gets BarSerializer and BarSerializer for ker.serializer/value.serializer through default properties.
//binding "standard-out" gets BarSerializer and BarSerializer for ker.serializer/value.serializer through default properties.
assertThat(customKafkaProducerProperties.getConfiguration().get("key.serializer")).isEqualTo("BarSerializer.class");
assertThat(customKafkaProducerProperties.getConfiguration().get("value.serializer")).isEqualTo("BarSerializer.class");
@@ -121,7 +116,7 @@ public class KafkaBinderExtendedPropertiesTest {
KafkaConsumerProperties customKafkaConsumerProperties =
(KafkaConsumerProperties)((ExtendedPropertiesBinder) kafkaBinder).getExtendedConsumerProperties("custom-in");
//binding "input" gets BarSerializer and BarSerializer for ker.serializer/value.serializer through default properties.
//binding "standard-in" gets BarSerializer and BarSerializer for ker.serializer/value.serializer through default properties.
assertThat(customKafkaConsumerProperties.getConfiguration().get("key.serializer")).isEqualTo("BarSerializer.class");
assertThat(customKafkaConsumerProperties.getConfiguration().get("value.serializer")).isEqualTo("BarSerializer.class");
@@ -133,8 +128,8 @@ public class KafkaBinderExtendedPropertiesTest {
@EnableAutoConfiguration
public static class KafkaMetricsTestConfig {
@StreamListener(Sink.INPUT)
@SendTo(Processor.OUTPUT)
@StreamListener("standard-in")
@SendTo("standard-out")
public String process(String payload) {
return payload;
}
@@ -147,14 +142,18 @@ public class KafkaBinderExtendedPropertiesTest {
}
interface CustomBindingForExtendedPropertyTesting extends Processor{
interface CustomBindingForExtendedPropertyTesting {
@Output("custom-out")
MessageChannel customOut();
@Input("standard-in")
SubscribableChannel standardIn();
@Output("standard-out")
MessageChannel standardOut();
@Input("custom-in")
SubscribableChannel customIn();
@Output("custom-out")
MessageChannel customOut();
}
}