Kafka auth

streampipes-backend/src/main/resources/shiro.ini

@@ -71,5 +71,6 @@ securityManager.rememberMeManager.cookie.maxAge = 1000000000
 /api/v2/connect/*/master/sources/* = anon
 /api/v2/connect/*/master/sources/*/streams = anon
 /api/v2/connect/*/master/sources/*/streams/* = anon
+/api/v2/connect/*/master/resolvable/*/configurations = anon
 /api/** = customFilter
 /** = customFilter

streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaConsumer.java

@@ -18,10 +18,7 @@
 
 package org.apache.streampipes.messaging.kafka;
 
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,6 +42,8 @@ public class SpKafkaConsumer implements EventConsumer<KafkaTransportProtocol>, R
   private String topic;
   private InternalEventProcessor<byte[]> eventProcessor;
   private KafkaTransportProtocol protocol;
+  private String username;
+  private String password;
   private volatile boolean isRunning;
   private Boolean patternTopic = false;
 
@@ -56,12 +55,25 @@ public SpKafkaConsumer() {
 
   public SpKafkaConsumer(KafkaTransportProtocol protocol, String topic, InternalEventProcessor<byte[]> eventProcessor) {
       this.protocol = protocol;
+      this.username = null;
+      this.password = null;
       this.topic = topic;
       this.eventProcessor = eventProcessor;
       this.isRunning = true;
   }
 
 
+  public SpKafkaConsumer(KafkaTransportProtocol protocol, String topic, InternalEventProcessor<byte[]> eventProcessor,
+                         String username, String password) {
+    this.protocol = protocol;
+    this.topic = topic;
+    this.eventProcessor = eventProcessor;
+    this.isRunning = true;
+    this.username = username;
+    this.password = password;
+  }
+
+
   // TODO backwards compatibility, remove later
   public SpKafkaConsumer(String kafkaUrl, String topic, InternalEventProcessor<byte[]> callback) {
     KafkaTransportProtocol protocol = new KafkaTransportProtocol();
@@ -78,12 +90,19 @@ public SpKafkaConsumer(String kafkaUrl, String topic, InternalEventProcessor<byt
 
   @Override
   public void run() {
-    KafkaConsumer<String, byte[]> kafkaConsumer = new KafkaConsumer<>(getProperties());
+    Properties props;
+    if (username != null && password != null) {
+      props = makePropertiesSaslPlain(protocol, username, password);
+    }
+    else {
+      props = makeProperties(protocol);
+    }
+    KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
     if (!patternTopic) {
-      kafkaConsumer.subscribe(Collections.singletonList(topic));
+      consumer.subscribe(Collections.singletonList(topic));
     } else {
       topic = replaceWildcardWithPatternFormat(topic);
-      kafkaConsumer.subscribe(Pattern.compile(topic), new ConsumerRebalanceListener() {
+      consumer.subscribe(Pattern.compile(topic), new ConsumerRebalanceListener() {
         @Override
         public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
           // TODO
@@ -96,24 +115,28 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
       });
     }
     while (isRunning) {
-      ConsumerRecords<String, byte[]> records = kafkaConsumer.poll(100);
+      ConsumerRecords<String, byte[]> records = consumer.poll(100);
       for (ConsumerRecord<String, byte[]> record : records) {
         eventProcessor.onEvent(record.value());
       }
     }
     LOG.info("Closing Kafka Consumer.");
-    kafkaConsumer.close();
+    consumer.close();
   }
 
   private String replaceWildcardWithPatternFormat(String topic) {
     topic = topic.replaceAll("\\.", "\\\\.");
     return topic.replaceAll("\\*", ".*");
   }
 
-  private Properties getProperties() {
+  private Properties makeProperties(KafkaTransportProtocol protocol) {
     return new ConsumerConfigFactory(protocol).makeProperties();
   }
 
+  private Properties makePropertiesSaslPlain(KafkaTransportProtocol protocol, String username, String password) {
+    return new ConsumerConfigFactory(protocol).makePropertiesSaslPlain(username, password);
+  }
+
   @Override
   public void connect(KafkaTransportProtocol protocol, InternalEventProcessor<byte[]>
           eventProcessor)

streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaProducer.java

@@ -52,9 +52,7 @@ public class SpKafkaProducer implements EventProducer<KafkaTransportProtocol>, S
 
   private static final Logger LOG = LoggerFactory.getLogger(SpKafkaProducer.class);
 
-  public SpKafkaProducer() {
-
-  }
+  public SpKafkaProducer() { }
 
   // TODO backwards compatibility, remove later
   public SpKafkaProducer(String url, String topic) {
@@ -66,6 +64,16 @@ public SpKafkaProducer(String url, String topic) {
     this.producer = new KafkaProducer<>(makeProperties(protocol));
   }
 
+  // TODO backwards compatibility, remove later
+  public SpKafkaProducer(String url, String topic, String username, String password) {
+    String[] urlParts = url.split(COLON);
+    KafkaTransportProtocol protocol = new KafkaTransportProtocol(urlParts[0],
+            Integer.parseInt(urlParts[1]), topic);
+    this.brokerUrl = url;
+    this.topic = topic;
+    this.producer = new KafkaProducer<>(makePropertiesSaslPlain(protocol, username, password));
+  }
+
   public void publish(String message) {
     publish(message.getBytes());
   }
@@ -78,6 +86,10 @@ private Properties makeProperties(KafkaTransportProtocol protocol) {
     return new ProducerConfigFactory(protocol).makeProperties();
   }
 
+  private Properties makePropertiesSaslPlain(KafkaTransportProtocol protocol, String username, String password) {
+    return new ProducerConfigFactory(protocol).makePropertiesSaslPlain(username, password);
+  }
+
   @Override
   public void connect(KafkaTransportProtocol protocol) {
     LOG.info("Kafka producer: Connecting to " + protocol.getTopicDefinition().getActualTopicName());

streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/config/AbstractConfigFactory.java

@@ -34,6 +34,8 @@ public AbstractConfigFactory(KafkaTransportProtocol protocol) {
 
   public abstract Properties makeProperties();
 
+  public abstract Properties makePropertiesSaslPlain(String username, String password);
+
   protected <T> T getConfigOrDefault(Supplier<T> function,
                                       T defaultValue) {
     return function.get() != null ? function.get() : defaultValue;

streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/config/ConsumerConfigFactory.java

@@ -17,7 +17,11 @@
  */
 package org.apache.streampipes.messaging.kafka.config;
 
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.KafkaClient;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
 
 import java.util.Properties;
@@ -33,6 +37,7 @@ public class ConsumerConfigFactory extends AbstractConfigFactory {
           ".serialization.StringDeserializer";
   private static final String VALUE_DESERIALIZER_CLASS_CONFIG_DEFAULT = "org.apache.kafka.common" +
           ".serialization.ByteArrayDeserializer";
+  private static final String SASL_MECHANISM = "PLAIN";
 
   public ConsumerConfigFactory(KafkaTransportProtocol protocol) {
     super(protocol);
@@ -41,7 +46,6 @@ public ConsumerConfigFactory(KafkaTransportProtocol protocol) {
   @Override
   public Properties makeProperties() {
     Properties props = new Properties();
-
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, getBrokerUrl());
     props.put(ConsumerConfig.GROUP_ID_CONFIG, getConfigOrDefault(protocol::getGroupId,
             UUID.randomUUID().toString()));
@@ -54,6 +58,17 @@ public Properties makeProperties() {
     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KEY_DESERIALIZER_CLASS_CONFIG_DEFAULT);
     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, VALUE_DESERIALIZER_CLASS_CONFIG_DEFAULT);
     props.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());
+
+    return props;
+  }
+
+  @Override
+  public Properties makePropertiesSaslPlain(String username, String password) {
+    Properties props = makeProperties();
+    props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.toString());
+    String SASL_JAAS_CONFIG = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";";
+    props.put(SaslConfigs.SASL_JAAS_CONFIG, SASL_JAAS_CONFIG);
     return props;
   }
 }

streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/config/ProducerConfigFactory.java

@@ -17,9 +17,16 @@
  */
 package org.apache.streampipes.messaging.kafka.config;
 
+import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator;
+import org.apache.kafka.common.security.scram.internals.ScramMechanism;
 import org.apache.streampipes.model.grounding.KafkaTransportProtocol;
 
+import javax.management.monitor.CounterMonitor;
 import java.util.Properties;
 
 public class ProducerConfigFactory extends AbstractConfigFactory {
@@ -59,4 +66,14 @@ public Properties makeProperties() {
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, VALUE_SERIALIZER_DEFAULT);
     return props;
   }
+
+  @Override
+  public Properties makePropertiesSaslPlain(String username, String password) {
+    Properties props = makeProperties();
+    props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+    props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_PLAINTEXT.toString());
+    String SASL_JAAS_CONFIG = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";";
+    props.put(SaslConfigs.SASL_JAAS_CONFIG, SASL_JAAS_CONFIG);
+    return props;
+  }
 }

streampipes-wrapper-python/streampipes/model/__init__.py

@@ -13,4 +13,4 @@
 # 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.
-#
+#

streampipes-wrapper-python/streampipes/utils/__init__.py

@@ -13,4 +13,4 @@
 # 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.
-#
+#

ui/src/app/core-ui/static-properties/static-runtime-resolvable-input/runtime-resolvable.service.ts

@@ -37,7 +37,7 @@ export class RuntimeResolvableService {
   }
 
   fetchRemoteOptionsForAdapter(resolvableOptionsParameterRequest: RuntimeOptionsRequest, adapterId: string): Observable<RuntimeOptionsResponse> {
-    let url: string = "/streampipes-connect/api/v1/"
+    let url: string = "/streampipes-backend/api/v2/connect/"
         + this.authStatusService.email
         + "/master/resolvable/"
         + encodeURIComponent(adapterId)