From 9443c621b0b586e9e4edc1a7753f7cef07bd13a6 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 10 Oct 2018 14:53:13 -0400 Subject: [PATCH] Kafka binder JAAS config refactoring (#461) * 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 --- pom.xml | 2 +- .../JaasLoginModuleConfiguration.java | 28 +--- .../config/KafkaBinderConfiguration.java | 17 +-- ...afkaBinderJaasInitializerListenerTest.java | 122 ++++++++++++++++++ .../KafkaBinderExtendedPropertiesTest.java | 39 +++--- 5 files changed, 149 insertions(+), 59 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java diff --git a/pom.xml b/pom.xml index a8a4c622f..6e293443a 100644 --- a/pom.xml +++ b/pom.xml @@ -12,7 +12,7 @@ 1.8 - 2.2.0.RC1 + 2.2.0.BUILD-SNAPSHOT 3.1.0.M1 2.0.0 2.1.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java index f82ec51f3..aa4e4755c 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/JaasLoginModuleConfiguration.java @@ -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 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 getOptions() { diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index c20068b2b..d562e0d85 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -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); } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java new file mode 100644 index 000000000..871a20de0 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java @@ -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 { + + } +} diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java index c55551be0..ec64de001 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java @@ -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(); } - }