Repository: kafka-ops/kafka-topology-builder (java)

Issue:
KTB are not providing correct ACLs for streams apps
KTB fails to provide correct ACLs for streams apps. According to [the documentation](https://docs.confluent.io/current/streams/developer-guide/security.html#required-acl-setting-for-secure-ak-clusters
) the following ACLs are needed:
- Topic resource (for internal topics): READ, DELETE, WRITE, CREATE (prefixed)
  - Or the ALL operation as KTB uses today
- Consumer Group resource: READ, DESCRIBE
- Topic resource (for input topic): READ
- Topic resource (for output topic): WRITE

KTB does not provide ACLs for the consumer group. Also for the internal topics the ALL operation is granted, instead more restricted operations should be granted.

Another problem is related to the feature for custom topic naming in KTB. This allows for topic naming with just the name given for the topic in yaml, then no topic prefix exists. So KTB must support some other way to give the prefix in these cases. Ref. the documentation linked above then the option is to provide the applicationId for the streams app.

Currently the following ACL is added in KTB to support streams app internal topics:
```
'TOPIC', <topicPrefix>, '*', 'ALL', 'User:StreamsUser', 'PREFIX'
```

When an applicationId is given for the streams app in the topology this should changed to:

```
'TOPIC', <streamApplicationId>, '*', 'ALL', 'User:StreamsUser', 'PREFIX'
```

The topology might then look like this: the `streams` section has a new key `applicationId`.  E.g.:
```
projects:
  - name: "projectA"
    streams:
      - principal: "User:StreamsUser"
        applicationId: "aStreamsApp"
        topics:
          read:
            - "topicA"
            - "topicB"
          write:
            - "topicC"
            - "topicD"
```

Internally this field will then be used to create prefix-based ACLs for both consumer group and internal topics. 

Pull request description:
Add streams acls for consumer group and support for applicationId
This will fix #116.

In addition improved the acl documentation for consumers and producers.

Patch:
diff --git a/README.md b/README.md
index 1600913..dc46b54 100644
--- a/README.md
+++ b/README.md
@@ -162,7 +162,8 @@ projects:
       num.partitions: "3"
 ```
 
-more examples can be found at the [example/](example/) directory.
+more examples can be found at the [example/](example/) directory. 
+Also check the documentation on ACL and RBAC in [the docs](https://kafka-topology-builder.readthedocs.io/) for information on what ACLs are created.
 
 ## Interested in contributing back?
 
diff --git a/docs/futures/what-acl-management.rst b/docs/futures/what-acl-management.rst
index 6ee13a1..542eb06 100644
--- a/docs/futures/what-acl-management.rst
+++ b/docs/futures/what-acl-management.rst
@@ -5,14 +5,14 @@ ACLs
 -----------
 
 In the topology descriptor files users can create permissions for different types of applications, Consumers, Producers, Kafka streams apps or Kafka Connectors.
-With this roles, users can easy create the permissions that map directly to their needs.
+With this roles, users can easily create the permissions that map directly to their needs.
 
 Consumers
 ^^^^^^^^^^^
 
 As a user you can configure consumers for each project.
-Consumer have a principal and optionally a consumer group name, if the consumer group is by default defined as prefix ACL with "*".
-Users can customize this property by defining a *group* attribute for each consumer.
+Consumer have a principal and optionally a consumer group name. The consumer group ACL is by default defined for all groups ("*").
+Users can customize this ACL by defining a *group* attribute for each consumer.
 
 
 .. code-block:: YAML
@@ -40,7 +40,7 @@ Consumer definition with principal "User:App0" and without an specific consumer
 
 Consumer definition with principal "User:App0" and consumer group name "foo".
 
-In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair or rules the KTB offer the option
+In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair of rules the KTB offer the option
 to optimise the number of ACLs using prefixed rules.
 
 The optimised ACLs/RBAC can be enabled using the *topology.acls.optimized* configuration property.
@@ -61,16 +61,65 @@ Producers have a principal.
         producers:
           - principal: "User:App0"
 
-In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair or rules the KTB offer the option
+In the default mode the KTB will create dedicated ACL for each user and topic pair. For organisations that aim not to have dedicated pair of rules the KTB offer the option
 to optimise the number of ACLs using prefixed rules.
 
 The optimised ACLs/RBAC can be enabled using the *topology.acls.optimized* configuration property.
 
+Streams
+^^^^^^^^^^^
+
+Users can also setup Kafka Streams applications.
+Each one of them will be composed of a principal and a list of topics that this principal needs to read and write.
+The principal is the user used by the streams app to connect to Kafka. You can also optionally specify the *applicationId*.
+
+.. code-block:: YAML
+
+  ---
+    context: "context"
+    source: "source"
+    projects:
+      - name: "foo"
+        streams:
+          - principal: "User:App0"
+            topics:
+              read:
+                - "topicA"
+              write:
+                - "topicB"
+
+KTB will create the necessary ACLs for reading and writing topics, as well as ACLs needed by the app to create/manage internal topics.
+The ACLs for the consumer group and for internal topic creation are prefixed.
+The resource name (prefix) is by default the topic name prefix in the project.
+For the example above the prefix will by default be "context.source.foo".
+
+As you see in the next example this can be overridden by specifying an *applicationId* in the topology.
+
+.. code-block:: YAML
+
+  ---
+    context: "context"
+    source: "source"
+    projects:
+      - name: "foo"
+        streams:
+          - principal: "User:App0"
+            applicationId: "streamsApplicationId"
+            topics:
+              read:
+                - "topicA"
+              write:
+                - "topicB"
+
+
+When the *applicationId* is specified this is used as the resource prefix in the ACLs for consumer groups
+and internal topics for the streams app. In the above example the prefix will be "streamsApplicationId".
+
 Connectors
 ^^^^^^^^^^^
 
 In a similar fashion as with the previous roles, users can setup specific Kafka Connect setups.
-Each one of them will be composed of a principal, this would be the user used by the connect to
+Each one of them will be composed of a principal, this would be the user used by the connector to
 connect to Kafka and a list of topics that this principal needs to read or write to, remember
 Connectors can either read (Sink) or write (Source) into Apache Kafka and they do it to topics.
 
@@ -133,13 +182,16 @@ What ACLs are created
 ^^^^^^^^^^^^^^^^^^^^^
 Kafka Topology Builder will assign the following ACLs:
 
-* each principal in the `consumers` list will get `READ` and `DESCRIBE` permissions on each topic in the containing project as well as `READ` access on every consumer group
-* each principal in the `producers` list will get `WRITE` and `DESCRIBE` permissions on each topic in the containing project
+* each principal in the `consumers` list will get `READ` and `DESCRIBE` permissions on each topic. In addition `READ` access on every consumer group (by default) or the group specified in the topology.
+
+* each principal in the `producers` list will get `WRITE` and `DESCRIBE` permissions on each topic. In addition if a *transactionId* is specified a WRITE and DESCRIBE ACL is created on the transactionId resource. And if either *transactionId* or *idempotence* is specified for the producer the IDEMPOTENT_WRITE ALLOW acl is created.
+
 * each principal in the `streams` list will get
 
   * `READ` access on every topic in its `read` sub-object
   * `WRITE` access on every topic `write` sub-object
-  * `ALL` access on every topic starting with fully-qualified project name, e.g. ``context.company.env.source.projectA`` in the example above. These are `PREFIXED` ACLs.
+  * `ALL` access on every topic starting with the fully-qualified project name (by default) or the given applicationId. These are `PREFIXED` ACLs.
+  * `READ` access on consumer groups starting with the fully-qualified project name (by default) or the given applicationId. These are `PREFIXED` ACLs.
 
 * each principal for a connector will get
 
diff --git a/example/descriptor.yaml b/example/descriptor.yaml
index 3a24795..06ae802 100644
--- a/example/descriptor.yaml
+++ b/example/descriptor.yaml
@@ -52,6 +52,24 @@ projects:
         config:
           replication.factor: "1"
           num.partitions: "1"
+  - name: "projectC"
+    streams:
+      - principal: "User:App0"
+        applicationId: "streamsApplicationId"
+        topics:
+          read:
+            - "topicE"
+          write:
+            - "topicF"
+    topics:
+      - name: "topicE"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "topicF"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
 platform:
   schema_registry:
     instances:
diff --git a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java
index 63a7b47..71a6d4e 100644
--- a/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java
+++ b/src/main/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreams.java
@@ -3,7 +3,6 @@ package com.purbon.kafka.topology.actions.access.builders;
 import com.purbon.kafka.topology.BindingsBuilderProvider;
 import com.purbon.kafka.topology.actions.BaseAccessControlAction;
 import com.purbon.kafka.topology.model.users.KStream;
-import java.io.IOException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -12,24 +11,24 @@ public class BuildBindingsForKStreams extends BaseAccessControlAction {
 
   private final BindingsBuilderProvider builderProvider;
   private final KStream app;
-  private final String topicPrefix;
+  private final String prefix;
 
   public BuildBindingsForKStreams(
       BindingsBuilderProvider builderProvider, KStream app, String topicPrefix) {
     super();
     this.builderProvider = builderProvider;
     this.app = app;
-    this.topicPrefix = topicPrefix;
+    this.prefix = app.getApplicationId().orElse(topicPrefix);
   }
 
   @Override
-  protected void execute() throws IOException {
+  protected void execute() {
     List<String> readTopics = app.getTopics().get(KStream.READ_TOPICS);
     List<String> writeTopics = app.getTopics().get(KStream.WRITE_TOPICS);
 
     bindings =
         builderProvider.buildBindingsForStreamsApp(
-            app.getPrincipal(), topicPrefix, readTopics, writeTopics);
+            app.getPrincipal(), prefix, readTopics, writeTopics);
   }
 
   @Override
@@ -37,7 +36,7 @@ public class BuildBindingsForKStreams extends BaseAccessControlAction {
     Map<String, Object> map = new HashMap<>();
     map.put("Operation", getClass().getName());
     map.put("Principal", app.getPrincipal());
-    map.put("Topic", topicPrefix);
+    map.put("Topic", prefix);
     return map;
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/model/users/KStream.java b/src/main/java/com/purbon/kafka/topology/model/users/KStream.java
index 8943068..7604d8e 100644
--- a/src/main/java/com/purbon/kafka/topology/model/users/KStream.java
+++ b/src/main/java/com/purbon/kafka/topology/model/users/KStream.java
@@ -1,16 +1,36 @@
 package com.purbon.kafka.topology.model.users;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
 import com.purbon.kafka.topology.model.DynamicUser;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
 
 public class KStream extends DynamicUser {
 
+  @JsonInclude(Include.NON_EMPTY)
+  private Optional<String> applicationId;
+
   public KStream() {
-    super();
+    this("", new HashMap<>());
   }
 
-  public KStream(String principal, HashMap<String, List<String>> topics) {
+  public KStream(
+      String principal, HashMap<String, List<String>> topics, Optional<String> applicationId) {
     super(principal, topics);
+    this.applicationId = applicationId;
+  }
+
+  public KStream(String principal, HashMap<String, List<String>> topics) {
+    this(principal, topics, Optional.empty());
+  }
+
+  public Optional<String> getApplicationId() {
+    return applicationId;
+  }
+
+  public void setApplicationId(Optional<String> applicationId) {
+    this.applicationId = applicationId;
   }
 }
diff --git a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
index aa657fc..f536c29 100644
--- a/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
+++ b/src/main/java/com/purbon/kafka/topology/roles/acls/AclsBindingsBuilder.java
@@ -82,8 +82,8 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
 
   @Override
   public List<TopologyAclBinding> buildBindingsForStreamsApp(
-      String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics) {
-    return toList(streamsAppStream(principal, topicPrefix, readTopics, writeTopics));
+      String principal, String prefix, List<String> readTopics, List<String> writeTopics) {
+    return toList(streamsAppStream(principal, prefix, readTopics, writeTopics));
   }
 
   @Override
@@ -165,7 +165,7 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
   }
 
   private Stream<AclBinding> streamsAppStream(
-      String principal, String topicPrefix, List<String> readTopics, List<String> writeTopics) {
+      String principal, String prefix, List<String> readTopics, List<String> writeTopics) {
 
     List<AclBinding> acls = new ArrayList<>();
 
@@ -178,7 +178,10 @@ public class AclsBindingsBuilder implements BindingsBuilderProvider {
             acls.add(
                 buildTopicLevelAcl(principal, topic, PatternType.LITERAL, AclOperation.WRITE)));
 
-    acls.add(buildTopicLevelAcl(principal, topicPrefix, PatternType.PREFIXED, AclOperation.ALL));
+    acls.add(buildTopicLevelAcl(principal, prefix, PatternType.PREFIXED, AclOperation.ALL));
+
+    acls.add(buildGroupLevelAcl(principal, prefix, PatternType.PREFIXED, AclOperation.READ));
+
     return acls.stream();
   }
 


Test patch:
diff --git a/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
index cdab754..065f180 100644
--- a/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
+++ b/src/test/java/com/purbon/kafka/topology/AclsBindingsBuilderTest.java
@@ -184,7 +184,7 @@ public class AclsBindingsBuilderTest {
         builder.buildBindingsForStreamsApp(
             stream.getPrincipal(), "prefix", readTopics, writeTopics);
 
-    assertThat(bindings.size()).isEqualTo(3);
+    assertThat(bindings.size()).isEqualTo(4);
     assertThat(bindings)
         .contains(
             buildTopicLevelAcl(
@@ -193,6 +193,10 @@ public class AclsBindingsBuilderTest {
         .contains(
             buildTopicLevelAcl(
                 stream.getPrincipal(), "bar", PatternType.LITERAL, AclOperation.WRITE));
+    assertThat(bindings)
+        .contains(
+            buildGroupLevelAcl(
+                stream.getPrincipal(), "prefix", PatternType.PREFIXED, AclOperation.READ));
     assertThat(bindings)
         .contains(
             buildTopicLevelAcl(
diff --git a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
index b47e3bc..5aef4e7 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologySerdesTest.java
@@ -201,6 +201,22 @@ public class TopologySerdesTest {
     assertThat(project.getProducers().get(2).getIdempotence()).isNotEmpty();
   }
 
+  @Test
+  public void testStreamsApps() throws IOException, URISyntaxException {
+
+    URL topologyDescriptor = getClass().getResource("/descriptor.yaml");
+    Topology topology = parser.deserialise(Paths.get(topologyDescriptor.toURI()).toFile());
+
+    Project project1 = topology.getProjects().get(0);
+    assertThat(project1.getStreams()).hasSize(1);
+    assertThat(project1.getStreams()).noneMatch(s -> s.getApplicationId().isPresent());
+
+    Project project3 = topology.getProjects().get(2);
+    assertThat(project3.getStreams()).hasSize(1);
+    assertThat(project3.getStreams())
+            .allMatch(s -> s.getApplicationId().orElse("notFound").equals("applicationId-1"));
+  }
+
   @Test(expected = IllegalArgumentException.class)
   public void testSchemaSerdes() throws URISyntaxException, IOException {
     URL topologyDescriptor = getClass().getResource("/descriptor-wrong-schemas.yaml");
@@ -208,7 +224,7 @@ public class TopologySerdesTest {
   }
 
   @Test
-  public void testPlaformProcessing() throws IOException, URISyntaxException {
+  public void testPlatformProcessing() throws IOException, URISyntaxException {
 
     URL topologyDescriptor = getClass().getResource("/descriptor.yaml");
 
diff --git a/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java b/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
index fe406a7..76c4f67 100644
--- a/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
+++ b/src/test/java/com/purbon/kafka/topology/TopologyValidationTest.java
@@ -87,11 +87,15 @@ public class TopologyValidationTest {
 
     TopologyValidator validator = new TopologyValidator(config);
     List<String> results = validator.validate(topology);
-    assertThat(results).hasSize(3);
+    assertThat(results).hasSize(5);
     assertThat(results.get(0)).isEqualTo("Project name does not follow the camelCase format: foo");
     assertThat(results.get(1))
         .isEqualTo("Topic contextOrg.source.foo.foo has an invalid number of partitions: 1");
     assertThat(results.get(2))
         .isEqualTo("Topic contextOrg.source.bar.bar.avro has an invalid number of partitions: 1");
+    assertThat(results.get(3))
+        .isEqualTo("Topic contextOrg.source.baz.topicE has an invalid number of partitions: 1");
+    assertThat(results.get(4))
+        .isEqualTo("Topic contextOrg.source.baz.topicF has an invalid number of partitions: 1");
   }
 }
diff --git a/src/test/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreamsTest.java b/src/test/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreamsTest.java
new file mode 100644
index 0000000..1d83a4f
--- /dev/null
+++ b/src/test/java/com/purbon/kafka/topology/actions/access/builders/BuildBindingsForKStreamsTest.java
@@ -0,0 +1,69 @@
+package com.purbon.kafka.topology.actions.access.builders;
+
+import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.purbon.kafka.topology.TopologyBuilderConfig;
+import com.purbon.kafka.topology.model.users.KStream;
+import com.purbon.kafka.topology.roles.acls.AclsBindingsBuilder;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+import org.apache.kafka.common.resource.ResourceType;
+import org.junit.Before;
+import org.junit.Test;
+
+public class BuildBindingsForKStreamsTest {
+
+  private BuildBindingsForKStreams action;
+  private AclsBindingsBuilder aclsBindingsBuilder;
+
+  @Before
+  public void setUp() {
+    aclsBindingsBuilder = new AclsBindingsBuilder(new TopologyBuilderConfig());
+  }
+
+  @Test
+  public void testStreamsWithoutApplicationId() {
+    HashMap<String, List<String>> topics = new HashMap<>();
+    topics.put(KStream.READ_TOPICS, singletonList("topicA"));
+    topics.put(KStream.WRITE_TOPICS, singletonList("topicB"));
+
+    KStream app = new KStream("User:user", topics);
+    String topicPrefix = "topicPrefix";
+    action = new BuildBindingsForKStreams(aclsBindingsBuilder, app, topicPrefix);
+    action.execute();
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.TOPIC
+                    && b.getResourceName().equals(topicPrefix));
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.GROUP
+                    && b.getResourceName().equals(topicPrefix));
+  }
+
+  @Test
+  public void testStreamWithApplicationId() {
+    HashMap<String, List<String>> topics = new HashMap<>();
+    topics.put(KStream.READ_TOPICS, singletonList("topicA"));
+    topics.put(KStream.WRITE_TOPICS, singletonList("topicB"));
+
+    String applicationId = "applicationId";
+    KStream app = new KStream("User:user", topics, Optional.of(applicationId));
+    action = new BuildBindingsForKStreams(aclsBindingsBuilder, app, "topicPrefix");
+    action.execute();
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.TOPIC
+                    && b.getResourceName().equals(applicationId));
+    assertThat(action.getBindings())
+        .anyMatch(
+            b ->
+                b.getResourceType() == ResourceType.GROUP
+                    && b.getResourceName().equals(applicationId));
+  }
+}
diff --git a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
index 96f5360..0465a23 100644
--- a/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
+++ b/src/test/java/com/purbon/kafka/topology/integration/AccessControlManagerIT.java
@@ -473,7 +473,7 @@ public class AccessControlManagerIT {
     acls = kafkaAdminClient.describeAcls(filter).values().get();
 
     // two acls created for the read topics
-    assertEquals(2, acls.size());
+    assertEquals(3, acls.size());
 
     entryFilter =
         new AccessControlEntryFilter(
diff --git a/src/test/resources/descriptor.yaml b/src/test/resources/descriptor.yaml
index 5692542..ad085c7 100644
--- a/src/test/resources/descriptor.yaml
+++ b/src/test/resources/descriptor.yaml
@@ -57,6 +57,24 @@ projects:
         config:
           replication.factor: "1"
           num.partitions: "1"
+  - name: "baz"
+    streams:
+      - principal: "User:App1"
+        applicationId: "applicationId-1"
+        topics:
+          read:
+            - "topicE"
+          write:
+            - "topicF"
+    topics:
+      - name: "topicE"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
+      - name: "topicF"
+        config:
+          replication.factor: "1"
+          num.partitions: "1"
 platform:
   schema_registry:
     instances: