METRON-1394 Create Rest endpoint to add the ACL for current user to kafka 
topics  (MohanDV via merrimanr) closes apache/metron#895


Project: http://git-wip-us.apache.org/repos/asf/metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/e265b369
Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/e265b369
Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/e265b369

Branch: refs/heads/feature/METRON-1344-test-infrastructure
Commit: e265b369192613335335514bc842f6aebe016d3e
Parents: 5f08ba0
Author: MohanDV <mohan...@gmail.com>
Authored: Wed Feb 21 08:32:01 2018 -0600
Committer: merrimanr <merrim...@apache.org>
Committed: Wed Feb 21 08:32:01 2018 -0600

----------------------------------------------------------------------
 .../metron/rest/controller/KafkaController.java |  1 +
 .../metron/rest/service/KafkaService.java       |  9 ++++++
 .../rest/service/impl/KafkaServiceImpl.java     | 33 ++++++++++++++++++--
 .../rest/service/impl/KafkaServiceImplTest.java | 10 ++++++
 4 files changed, 51 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/metron/blob/e265b369/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/controller/KafkaController.java
----------------------------------------------------------------------
diff --git 
a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/controller/KafkaController.java
 
b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/controller/KafkaController.java
index d057ac4..d04e227 100644
--- 
a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/controller/KafkaController.java
+++ 
b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/controller/KafkaController.java
@@ -119,4 +119,5 @@ public class KafkaController {
     kafkaService.produceMessage(name, message);
     return new ResponseEntity<>(HttpStatus.OK);
   }
+
 }

http://git-wip-us.apache.org/repos/asf/metron/blob/e265b369/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/KafkaService.java
----------------------------------------------------------------------
diff --git 
a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/KafkaService.java
 
b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/KafkaService.java
index da3b226..aa35c5d 100644
--- 
a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/KafkaService.java
+++ 
b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/KafkaService.java
@@ -68,4 +68,13 @@ public interface KafkaService {
   String getSampleMessage(String topic);
 
   void produceMessage(String topic, String message) throws RestException;
+
+
+  /**
+   *
+   * @param name The name of the Kafka topic to add the ACL.
+   * @return If topic was present true; otherwise false.
+   */
+  boolean addACLToCurrentUser(String name);
+
 }

http://git-wip-us.apache.org/repos/asf/metron/blob/e265b369/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/KafkaServiceImpl.java
----------------------------------------------------------------------
diff --git 
a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/KafkaServiceImpl.java
 
b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/KafkaServiceImpl.java
index 4f232fb..ac001b5 100644
--- 
a/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/KafkaServiceImpl.java
+++ 
b/metron-interface/metron-rest/src/main/java/org/apache/metron/rest/service/impl/KafkaServiceImpl.java
@@ -17,11 +17,13 @@
  */
 package org.apache.metron.rest.service.impl;
 
-import java.util.HashSet;
+import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
+import java.util.HashSet;
+import java.util.Map;
 import java.util.stream.Collectors;
+import kafka.admin.AclCommand;
 import kafka.admin.AdminOperationException;
 import kafka.admin.AdminUtils$;
 import kafka.admin.RackAwareMode;
@@ -33,12 +35,15 @@ import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.metron.rest.MetronRestConstants;
 import org.apache.metron.rest.RestException;
 import org.apache.metron.rest.model.KafkaTopic;
 import org.apache.metron.rest.service.KafkaService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.env.Environment;
 import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.security.core.userdetails.User;
 import org.springframework.stereotype.Service;
 
 /**
@@ -83,6 +88,9 @@ public class KafkaServiceImpl implements KafkaService {
     if (!listTopics().contains(topic.getName())) {
       try {
         adminUtils.createTopic(zkUtils, topic.getName(), 
topic.getNumPartitions(), topic.getReplicationFactor(), topic.getProperties(), 
RackAwareMode.Disabled$.MODULE$);
+        if 
(environment.getProperty(MetronRestConstants.KERBEROS_ENABLED_SPRING_PROPERTY, 
Boolean.class, false)){
+          addACLToCurrentUser(topic.getName());
+        }
       } catch (AdminOperationException e) {
         throw new RestException(e);
       }
@@ -154,4 +162,25 @@ public class KafkaServiceImpl implements KafkaService {
   public void produceMessage(String topic, String message) throws 
RestException {
     kafkaProducer.send(new ProducerRecord<>(topic, message));
   }
+
+  @Override
+  public boolean addACLToCurrentUser(String name){
+    if(listTopics().contains(name)) {
+      String zkServers = 
environment.getProperty(MetronRestConstants.ZK_URL_SPRING_PROPERTY);
+      User principal = (User) 
SecurityContextHolder.getContext().getAuthentication().getPrincipal();
+      String user = principal.getUsername();
+      List<String> cmd = new ArrayList<>();
+      cmd.add("--add");
+      cmd.add("--allow-principal");
+      cmd.add("User:" + user);
+      cmd.add("--topic");
+      cmd.add(name);
+      cmd.add("--authorizer-properties");
+      cmd.add("zookeeper.connect=" + String.join(",", zkServers));
+      AclCommand.main(cmd.toArray(new String[cmd.size()]));
+    } else {
+      return false;
+    }
+    return true;
+  }
 }

http://git-wip-us.apache.org/repos/asf/metron/blob/e265b369/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/service/impl/KafkaServiceImplTest.java
----------------------------------------------------------------------
diff --git 
a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/service/impl/KafkaServiceImplTest.java
 
b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/service/impl/KafkaServiceImplTest.java
index 4527e8e..b99128a 100644
--- 
a/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/service/impl/KafkaServiceImplTest.java
+++ 
b/metron-interface/metron-rest/src/test/java/org/apache/metron/rest/service/impl/KafkaServiceImplTest.java
@@ -63,6 +63,7 @@ import 
org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 import org.springframework.kafka.core.ConsumerFactory;
 
+
 @SuppressWarnings("unchecked")
 @RunWith(PowerMockRunner.class)
 @PowerMockIgnore("javax.management.*") // resolve classloader conflict
@@ -79,6 +80,7 @@ public class KafkaServiceImplTest {
 
   private KafkaService kafkaService;
 
+
   private static final KafkaTopic VALID_KAFKA_TOPIC = new KafkaTopic() {{
     setReplicationFactor(2);
     setNumPartitions(1);
@@ -314,4 +316,12 @@ public class KafkaServiceImplTest {
     verify(kafkaProducer).send(new ProducerRecord<>(topicName, 
expectedMessage));
     verifyZeroInteractions(kafkaProducer);
   }
+
+  @Test
+  public void addACLtoNonExistingTopicShouldReturnFalse() throws Exception{
+    when(kafkaConsumer.listTopics()).thenReturn(Maps.newHashMap());
+    assertFalse(kafkaService.addACLToCurrentUser("non_existent_topic"));
+  }
+
+
 }

Reply via email to