Repository: kafka-ops/kafka-topology-builder
Pull request #124 (2020-11-05 23:43:33)

Add topic level Access Control Rules (ACLs/RBAC)

purbon: Awesome! thanks a lot @michaelpearce-gain , appreciate a ton! can you add as well a few test for this new functionality? I only see basic ones for the serdes.

That would be really awesome to have this onboard.
michaelpearce-gain: Can you point me to existing tests for acls, i couldnt find.
purbon: here you have.
https://github.com/kafka-ops/kafka-topology-builder/blob/master/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java

https://github.com/kafka-ops/kafka-topology-builder/blob/master/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
purbon: Hi @michaelandrepearce and @michaelpearce-gain I really would like to merge this in, but test are missing for now. Do you mind If I add it and then we can move into merging them? 
michaelpearce-gain: @purbon feel free, i simply havent had a spare moment still.
purbon: Thanks, @michaelandrepearce, I took the time to add some tests and docs here! I will merge! 

thanks 

Patch:
diff --git a/docs/futures/what-topic-management.rst b/docs/futures/what-topic-management.rst
index 65fb791..baeb604 100644
--- a/docs/futures/what-topic-management.rst
+++ b/docs/futures/what-topic-management.rst
@@ -58,6 +58,30 @@ In following configuration the reader is seeing an example of how to create a to
 
 Users can as well increase later the number of partitions and the Topology Builder will handle it properly.
 
+Topic Level Consumers and Producers
+-----------
+
+It is possible to setup dedicated access control rules for specific topics instead of project scope.
+An example configuration would look like this:
+
+.. code-block:: YAML
+
+  ---
+    context: "context"
+    projects:
+      - name: "projectA"
+        topics:
+          - name: "foo"
+            consumers:
+              - principal: "User:App0"
+            producers:
+              - principal: "User:App1"
+            config:
+              replication.factor: "1"
+              num.partitions: "1"
+
+This type of Access Control rules allow the reader to setup dedicated access to single topics, without giving global project access.
+
 Handling Configuration
 -----------
 
diff --git a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
index a328b28..084e1a7 100644
--- a/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
+++ b/src/main/java/com/purbon/kafka/topology/AccessControlManager.java
@@ -16,13 +16,16 @@ import com.purbon.kafka.topology.model.Project;
 import com.purbon.kafka.topology.model.Topology;
 import com.purbon.kafka.topology.model.User;
 import com.purbon.kafka.topology.model.users.Connector;
+import com.purbon.kafka.topology.model.users.Consumer;
 import com.purbon.kafka.topology.model.users.KStream;
+import com.purbon.kafka.topology.model.users.Producer;
 import com.purbon.kafka.topology.model.users.Schemas;
 import com.purbon.kafka.topology.model.users.platform.ControlCenterInstance;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
 import com.purbon.kafka.topology.roles.TopologyAclBinding;
 import java.io.PrintStream;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -116,26 +119,41 @@ public class AccessControlManager {
     actions.add(
         new BuildBindingsForProducer(
             bindingsBuilder, project.getProducers(), project.namePrefix(), true));
+
+    // When optimised, still need to add any topic level specific.
+    actions.addAll(buildBasicUsersAcls(project, false));
     return actions;
   }
 
   private List<Action> buildDetailedConsumerAndProducerAcls(Project project) {
+    return buildBasicUsersAcls(project, true);
+  }
+
+  private List<Action> buildBasicUsersAcls(Project project, boolean includeProjectLevel) {
     List<Action> actions = new ArrayList<>();
     project
         .getTopics()
         .forEach(
             topic -> {
               final String fullTopicName = topic.toString();
-              if (!project.getConsumers().isEmpty()) {
+              Set<Consumer> consumers = new HashSet<>(topic.getConsumers());
+              if (includeProjectLevel) {
+                consumers.addAll(project.getConsumers());
+              }
+              if (!consumers.isEmpty()) {
                 Action action =
                     new BuildBindingsForConsumer(
-                        bindingsBuilder, project.getConsumers(), fullTopicName, false);
+                        bindingsBuilder, new ArrayList<>(consumers), fullTopicName, false);
                 actions.add(action);
               }
-              if (!project.getProducers().isEmpty()) {
+              Set<Producer> producers = new HashSet<>(topic.getProducers());
+              if (includeProjectLevel) {
+                producers.addAll(project.getProducers());
+              }
+              if (!producers.isEmpty()) {
                 Action action =
                     new BuildBindingsForProducer(
-                        bindingsBuilder, project.getProducers(), fullTopicName, false);
+                        bindingsBuilder, new ArrayList<>(producers), fullTopicName, false);
                 actions.add(action);
               }
             });
diff --git a/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java b/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
index 2d3e3c6..46be415 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Impl/TopicImpl.java
@@ -7,8 +7,12 @@ import com.purbon.kafka.topology.TopicManager;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.TopicSchemas;
+import com.purbon.kafka.topology.model.users.Consumer;
+import com.purbon.kafka.topology.model.users.Producer;
 import com.purbon.kafka.topology.utils.JinjaUtils;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
@@ -24,6 +28,9 @@ public class TopicImpl implements Topic, Cloneable {
 
   private String name;
 
+  private List<Producer> producers;
+  private List<Consumer> consumers;
+
   private Map<String, String> config;
   @JsonIgnore private TopologyBuilderConfig appConfig;
   @JsonIgnore private Map<String, Object> context;
@@ -64,7 +71,19 @@ public class TopicImpl implements Topic, Cloneable {
       Optional<String> dataType,
       Map<String, String> config,
       TopologyBuilderConfig appConfig) {
+    this(name, new ArrayList<>(), new ArrayList<>(), dataType, config, appConfig);
+  }
+
+  public TopicImpl(
+      String name,
+      List<Producer> producers,
+      List<Consumer> consumers,
+      Optional<String> dataType,
+      Map<String, String> config,
+      TopologyBuilderConfig appConfig) {
     this.name = name;
+    this.producers = producers;
+    this.consumers = consumers;
     this.dataType = dataType;
     this.config = config;
     this.replicationFactor = 0;
@@ -151,6 +170,28 @@ public class TopicImpl implements Topic, Cloneable {
     return replicationFactor;
   }
 
+  @Override
+  public List<Consumer> getConsumers() {
+    return consumers;
+  }
+
+  @Override
+  public List<Producer> getProducers() {
+    return producers;
+  }
+
+  @Override
+  public void setConsumers(List<Consumer> consumers) {
+    this.consumers.clear();
+    this.consumers.addAll(consumers);
+  }
+
+  @Override
+  public void setProducers(List<Producer> producers) {
+    this.producers.clear();
+    this.producers.addAll(producers);
+  }
+
   @Override
   public int partitionsCount() {
     return partitionCount;
@@ -169,7 +210,8 @@ public class TopicImpl implements Topic, Cloneable {
     try {
       return (Topic) super.clone();
     } catch (CloneNotSupportedException e) {
-      return new TopicImpl(getName(), getDataType(), getConfig(), getAppConfig());
+      return new TopicImpl(
+          getName(), getProducers(), getConsumers(), getDataType(), getConfig(), getAppConfig());
     }
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/model/Topic.java b/src/main/java/com/purbon/kafka/topology/model/Topic.java
index 6b849d0..1833944 100644
--- a/src/main/java/com/purbon/kafka/topology/model/Topic.java
+++ b/src/main/java/com/purbon/kafka/topology/model/Topic.java
@@ -3,7 +3,10 @@ package com.purbon.kafka.topology.model;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
+import com.purbon.kafka.topology.model.users.Consumer;
+import com.purbon.kafka.topology.model.users.Producer;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
@@ -31,4 +34,12 @@ public interface Topic {
   void setPrefixContext(Map<String, Object> prefixContext);
 
   short replicationFactor();
+
+  List<Consumer> getConsumers();
+
+  List<Producer> getProducers();
+
+  void setConsumers(List<Consumer> consumers);
+
+  void setProducers(List<Producer> producers);
 }
diff --git a/src/main/java/com/purbon/kafka/topology/model/users/Producer.java b/src/main/java/com/purbon/kafka/topology/model/users/Producer.java
index 9479985..5f9bdd3 100644
--- a/src/main/java/com/purbon/kafka/topology/model/users/Producer.java
+++ b/src/main/java/com/purbon/kafka/topology/model/users/Producer.java
@@ -1,6 +1,7 @@
 package com.purbon.kafka.topology.model.users;
 
 import com.purbon.kafka.topology.model.User;
+import java.util.Objects;
 import java.util.Optional;
 
 public class Producer extends User {
@@ -35,4 +36,21 @@ public class Producer extends User {
   public void setIdempotence(Optional<Boolean> idempotence) {
     this.idempotence = idempotence;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof Producer)) {
+      return false;
+    }
+    Producer producer = (Producer) o;
+    return getPrincipal().equals(producer.getPrincipal());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(getPrincipal());
+  }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
index 78460d7..cbb7b64 100644
--- a/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
+++ b/src/main/java/com/purbon/kafka/topology/serdes/TopicCustomDeserializer.java
@@ -10,8 +10,13 @@ import com.google.common.collect.Maps;
 import com.purbon.kafka.topology.TopologyBuilderConfig;
 import com.purbon.kafka.topology.model.Impl.TopicImpl;
 import com.purbon.kafka.topology.model.TopicSchemas;
+import com.purbon.kafka.topology.model.User;
+import com.purbon.kafka.topology.model.users.Consumer;
+import com.purbon.kafka.topology.model.users.Producer;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import jersey.repackaged.com.google.common.collect.Sets;
@@ -40,6 +45,8 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
     validateRequiresKeys(rootNode, "name");
 
     String name = rootNode.get("name").asText();
+    List<Consumer> consumers = getUsers(parser, rootNode, "consumers", Consumer.class);
+    List<Producer> producers = getUsers(parser, rootNode, "producers", Producer.class);
 
     Optional<JsonNode> optionalDataTypeNode = Optional.ofNullable(rootNode.get("dataType"));
     Optional<String> optionalDataType = optionalDataTypeNode.map(o -> o.asText());
@@ -52,7 +59,8 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
                     Maps.asMap(Sets.newHashSet(node.fieldNames()), (key) -> node.get(key).asText()))
             .orElse(new HashMap<>());
 
-    TopicImpl topic = new TopicImpl(name, optionalDataType, config, this.config);
+    TopicImpl topic =
+        new TopicImpl(name, producers, consumers, optionalDataType, config, this.config);
 
     Optional.ofNullable(rootNode.get("schemas"))
         .ifPresent(
@@ -66,4 +74,13 @@ public class TopicCustomDeserializer extends StdDeserializer<TopicImpl> {
         String.format("Topic %s with config %s has been created", topic.getName(), config));
     return topic;
   }
+
+  private <T extends User> List<T> getUsers(
+      JsonParser parser, JsonNode rootNode, String fieldName, Class<T> tClass)
+      throws com.fasterxml.jackson.core.JsonProcessingException {
+    JsonNode jsonNode = rootNode.get(fieldName);
+    return jsonNode == null
+        ? new ArrayList<>()
+        : new JsonSerdesUtils<T>().parseApplicationUser(parser, jsonNode, tClass);
+  }
 }


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
index 444c9e2..fea1576 100644
--- a/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AccessControlManagerTest.java
@@ -3,9 +3,13 @@ package com.purbon.kafka.topology;
 import static com.purbon.kafka.topology.BuilderCLI.BROKERS_OPTION;
 import static com.purbon.kafka.topology.TopologyBuilderConfig.OPTIMIZED_ACLS_CONFIG;
 import static java.util.Arrays.asList;
+import static org.hamcrest.CoreMatchers.hasItem;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.refEq;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.times;
@@ -38,6 +42,7 @@ import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
 import java.io.IOException;
 import java.io.PrintStream;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -54,6 +59,7 @@ import org.apache.kafka.common.resource.ResourceType;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.Matchers;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnit;
@@ -138,6 +144,44 @@ public class AccessControlManagerTest {
         .buildBindingsForConsumers(eq(users), eq(project.namePrefix()), eq(true));
   }
 
+  @Test
+  public void testConsumerAclsAtTopicLevel() {
+
+    Consumer projectConsumer = new Consumer("project-consumer");
+    Consumer topicConsumer = new Consumer("topic-consumer");
+
+    List<Consumer> projectConsumers = Collections.singletonList(projectConsumer);
+
+    Topology topology = new TopologyImpl();
+    topology.setContext("testConsumerAclsAtTopicLevel");
+
+    Project project = new ProjectImpl("project");
+    project.setConsumers(projectConsumers);
+    Topic topic = new TopicImpl("foo");
+
+    List<Consumer> topicConsumers = Arrays.asList(projectConsumer, topicConsumer);
+    topic.setConsumers(topicConsumers);
+
+    project.addTopic(topic);
+    topology.addProject(project);
+
+    doReturn(new ArrayList<TopologyAclBinding>())
+        .when(aclsBuilder)
+        .buildBindingsForConsumers(any(), eq(topic.toString()), eq(false));
+
+    accessControlManager.apply(topology, plan);
+
+    ArgumentCaptor<List> argumentCaptor = ArgumentCaptor.forClass(List.class);
+
+    verify(aclsBuilder, times(1))
+        .buildBindingsForConsumers(argumentCaptor.capture(), eq(topic.toString()), eq(false));
+
+    List<Consumer> capturedList = argumentCaptor.getValue();
+    assertThat(capturedList, hasItem(projectConsumer));
+    assertThat(capturedList, hasItem(topicConsumer));
+    assertThat(capturedList, hasSize(2));
+  }
+
   @Test
   public void newProducerACLsCreation() {
 
@@ -191,7 +235,45 @@ public class AccessControlManagerTest {
   }
 
   @Test
-  public void newKafkaStreamsAppACLsCreation() throws IOException {
+  public void testProducerAclsAtTopicLevel() {
+
+    Producer projectProducer = new Producer("project-producer");
+    Producer topicProducer = new Producer("topic-producer");
+
+    List<Producer> projectProducers = Collections.singletonList(projectProducer);
+
+    Topology topology = new TopologyImpl();
+    topology.setContext("testProducerAclsAtTopicLevel");
+
+    Project project = new ProjectImpl("project");
+    project.setProducers(projectProducers);
+    Topic topic = new TopicImpl("foo");
+
+    List<Producer> topicProducers = Arrays.asList(projectProducer, topicProducer);
+    topic.setProducers(topicProducers);
+
+    project.addTopic(topic);
+    topology.addProject(project);
+
+    doReturn(new ArrayList<TopologyAclBinding>())
+        .when(aclsBuilder)
+        .buildBindingsForProducers(any(), eq(topic.toString()), eq(false));
+
+    accessControlManager.apply(topology, plan);
+
+    ArgumentCaptor<List> argumentCaptor = ArgumentCaptor.forClass(List.class);
+
+    verify(aclsBuilder, times(1))
+        .buildBindingsForProducers(argumentCaptor.capture(), eq(topic.toString()), eq(false));
+
+    List<Producer> capturedList = argumentCaptor.getValue();
+    assertThat(capturedList, hasItem(projectProducer));
+    assertThat(capturedList, hasItem(topicProducer));
+    assertThat(capturedList, hasSize(2));
+  }
+
+  @Test
+  public void newKafkaStreamsAppACLsCreation() {
 
     Project project = new ProjectImpl();
 
@@ -417,7 +499,8 @@ public class AccessControlManagerTest {
     accessControlManager.apply(topology, plan);
     plan.run();
 
-    verify(aclsBuilder, times(1)).buildBindingsForConsumers(consumers, topicA.toString(), false);
+    verify(aclsBuilder, times(1))
+        .buildBindingsForConsumers(refEq(consumers), eq(topicA.toString()), eq(false));
 
     consumers = new ArrayList<>();
     consumers.add(new Consumer("User:app1"));
@@ -425,7 +508,7 @@ public class AccessControlManagerTest {
     bindings = returnAclsForConsumers(consumers, topicA.getName());
     doReturn(bindings)
         .when(aclsBuilder)
-        .buildBindingsForConsumers(eq(consumers), eq(topicA.toString()), eq(false));
+        .buildBindingsForConsumers(any(), eq(topicA.toString()), eq(false));
 
     Topology newTopology = buildTopology(consumers, asList(topicA));
 
diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
index 19bf65b..beaf987 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -237,6 +237,30 @@ public class TopologySerdesTest {
     assertTrue(topology.getProjects().get(0).getZookeepers().isEmpty());
   }
 
+  @Test
+  public void testRBACTopics() throws URISyntaxException, IOException {
+
+    URL topologyDescriptor = getClass().getResource("/descriptor-with-rbac-topics.yaml");
+    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+
+    Project project = topology.getProjects().get(0);
+    assertEquals("contextOrg", topology.getContext());
+    assertThat(project.getConnectors()).isEmpty();
+    assertThat(project.getProducers()).isEmpty();
+    assertThat(project.getStreams()).isEmpty();
+    assertThat(project.getZookeepers()).isEmpty();
+
+    Topic topic = project.getTopics().get(0);
+
+    assertThat(topic.getConsumers()).hasSize(1);
+    assertThat(topic.getConsumers()).contains(new Consumer("User:App0"));
+
+    assertThat(topic.getProducers()).hasSize(1);
+    assertThat(topic.getProducers()).contains(new Producer("User:App1"));
+
+    assertEquals("foo", topic.getName());
+  }
+
   @Test
   public void testWithRBACDescriptor() throws IOException, URISyntaxException {
     URL descriptor = getClass().getResource("/descriptor-with-rbac.yaml");
diff --git a/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
index d6c80ad..9976664 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/TopicManagerIT.java
@@ -57,7 +57,9 @@ public class TopicManagerIT {
 
   @BeforeClass
   public static void setup() {
-    container = new KafkaContainer(TestcontainersConfiguration.getInstance().getKafkaDockerImageName().withTag("5.5.0"));
+    container =
+        new KafkaContainer(
+            TestcontainersConfiguration.getInstance().getKafkaDockerImageName().withTag("5.5.0"));
     container.start();
   }
 
diff --git a/src/test/resources/descriptor-with-rbac-topics.yaml b/src/test/resources/descriptor-with-rbac-topics.yaml
new file mode 100644
index 0000000..c5b6e87
--- /dev/null
+++ b/src/test/resources/descriptor-with-rbac-topics.yaml
@@ -0,0 +1,30 @@
+---
+context: "contextOrg"
+source: "source"
+projects:
+  - name: "foo"
+    topics:
+      - name: "foo"
+        consumers:
+          - principal: "User:App0"
+        producers:
+          - principal: "User:App1"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "bar"
+        dataType: "avro"
+        consumers:
+          - principal: "User:App3"
+        producers:
+          - principal: "User:App4"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+  - name: "bar"
+    topics:
+      - dataType: "avro"
+        name: "bar"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"