Add a possibility to add custom roles definitions for julieops (#336)

* initial step into adding custom roles for julieops

* add test and wire up transformations and actions for having custom role definitions in acls and rbac providers

* add integration test for rbac

* add connector level test for rbac

* add sample actions and tests for ksqldb

* simplify acl conversion

* add early validation for wrong ACLs config values

* add sample documentation page

* ammend rbac julieroles test to make it more repetable

* make role and operation be an alias

docs/futures/define-custom-roles.rst

@@ -0,0 +1,128 @@
+Define custom roles for JulieOps
+*******************************
+
+While JulieOps offer you as a user the possibility to manage the ACLs (and RBAC if you're using the Confluent Platform) for most common
+applications deployments such as Consumers, Producers, Kafka Streams, Connectors and ksqlDB, it would be for some cases amazing to be
+be able to keep using the powerful abstractions of JulieOps but provide your own set of ACLs.
+
+For example:
+
+* If you are deploying a custom App and aim to give application specific roles
+* Deploying applications that might not fit our of the box with generic permissions provided by JulieOps
+* Or just if you are building your own roles based on Simple ACLs or Confluent RBAC
+
+and more.
+
+But, how can you get this with JulieOps.
+
+Defining the your roles
+-----------
+
+First thing is to define your roles in a configuration file, this file should look like this:
+
+.. code-block:: YAML
+
+    roles:
+      - name: "app"
+        acls:
+          - resourceType: "Topic"
+            resourceName: "{{topic}}"
+            patternType: "PREFIXED"
+            host: "*"
+            operation: "ALL"
+            permissionType: "ALLOW"
+          - resourceType: "Topic"
+            resourceName: "sourceTopic"
+            patternType: "LITERAL"
+            host: "*"
+            operation: "ALL"
+            permissionType: "READ"
+          - resourceType: "Topic"
+            resourceName: "targetTopic"
+            patternType: "LITERAL"
+            host: "*"
+            operation: "ALL"
+            permissionType: "WRITE"
+          - resourceType: "Group"
+            resourceName: "{{group}}"
+            patternType: "PREFIXED"
+            host: "*"
+            operation: "READ"
+            permissionType: "ALLOW"
+
+if you are using Confluent Platform RBAC functionality to define your own Access Control management, the only different property
+per acl is **role**, so the file might look like this:
+
+.. code-block:: YAML
+
+    roles:
+      - name: "app"
+        acls:
+          - resourceType: "Topic"
+            resourceName: "{{topic}}"
+            patternType: "PREFIXED"
+            host: "*"
+            role: "ResourceOwner"
+          - resourceType: "Topic"
+            resourceName: "sourceTopic"
+            patternType: "LITERAL"
+            host: "*"
+            role: "DeveloperRead"
+          - resourceType: "Topic"
+            resourceName: "targetTopic"
+            patternType: "LITERAL"
+            host: "*"
+            role: "DeveloperWrite"
+          - resourceType: "Group"
+            resourceName: "{{group}}"
+            patternType: "PREFIXED"
+            host: "*"
+            role: "DeveloperRead"
+          - resourceType: "Subject"
+            resourceName: "Subject:foo"
+            patternType: "LITERAL"
+            host: "*"
+            role: "DeveloperRead"
+          - resourceType: "Connector"
+            resourceName: "Connector:con"
+            patternType: "LITERAL"
+            host: "*"
+            role: "SecurityAdmin"
+          - resourceType: "KsqlCluster"
+            resourceName: "KsqlCluster:ksql-cluster"
+            patternType: "LITERAL"
+            host: "*"
+            role: "ResourceOwner"
+
+
+Plug this into JulieOps
+-----------
+
+Once the roles are define, the only thing you need to do is to configure your deployment to use it. This can be done using this
+configuration variable in your property file:
+
+
+.. code-block:: JAVA
+
+    julie.roles=/path/to/the/roles/file
+
+
+How would my new topology file look like
+-----------
+
+Once the new roles are setup, your topology can start using them just as the previous "hardcoded" roles.
+Your topology file could look like this:
+
+
+.. code-block:: YAML
+
+    context: "contextOrg"
+    source: "source"
+    projects:
+      - name: "foo"
+        foo:
+          - principal: "User:banana"
+            group: "foo"
+        bar:
+          - principal: "User:bandana"
+            group: "bar"

example/roles.yml

@@ -0,0 +1,31 @@
+---
+roles:
+  rules:
+    - name: "security"
+      acls:
+        - resourceType: "Topic"
+          resourceName: "{{topic}}"
+          patternType: ""
+          host: ""
+          operation: ""
+          permissionType: ""
+        - resourceType: "Group"
+          resourceName: "{{group}}"
+          patternType: ""
+          host: ""
+          operation: ""
+          permissionType: ""
+    - name: "other"
+      acls:
+        - resourceType: "Topic"
+          resourceName: ""
+          patternType: ""
+          host: ""
+          operation: ""
+          permissionType: ""
+        - resourceType: "Group"
+          resourceName: ""
+          patternType: ""
+          host: ""
+          operation: ""
+          permissionType: ""

src/main/java/com/purbon/kafka/topology/AccessControlManager.java

@@ -9,6 +9,7 @@
 import com.purbon.kafka.topology.actions.access.builders.rbac.*;
 import com.purbon.kafka.topology.model.Component;
 import com.purbon.kafka.topology.model.DynamicUser;
+import com.purbon.kafka.topology.model.JulieRoles;
 import com.purbon.kafka.topology.model.Platform;
 import com.purbon.kafka.topology.model.Project;
 import com.purbon.kafka.topology.model.Topology;
@@ -17,6 +18,7 @@
 import com.purbon.kafka.topology.model.users.Consumer;
 import com.purbon.kafka.topology.model.users.KSqlApp;
 import com.purbon.kafka.topology.model.users.KStream;
+import com.purbon.kafka.topology.model.users.Other;
 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;
@@ -37,6 +39,7 @@ public class AccessControlManager {
   private static final Logger LOGGER = LogManager.getLogger(AccessControlManager.class);
 
   private final Configuration config;
+  private final JulieRoles julieRoles;
   private AccessControlProvider controlProvider;
   private BindingsBuilderProvider bindingsBuilder;
   private final List<String> managedServiceAccountPrefixes;
@@ -52,12 +55,21 @@ public AccessControlManager(
       AccessControlProvider controlProvider,
       BindingsBuilderProvider builderProvider,
       Configuration config) {
+    this(controlProvider, builderProvider, new JulieRoles(), config);
+  }
+
+  public AccessControlManager(
+      AccessControlProvider controlProvider,
+      BindingsBuilderProvider builderProvider,
+      JulieRoles julieRoles,
+      Configuration config) {
     this.controlProvider = controlProvider;
     this.bindingsBuilder = builderProvider;
     this.config = config;
     this.managedServiceAccountPrefixes = config.getServiceAccountManagedPrefixes();
     this.managedTopicPrefixes = config.getTopicManagedPrefixes();
     this.managedGroupPrefixes = config.getGroupManagedPrefixes();
+    this.julieRoles = julieRoles;
   }
 
   /**
@@ -68,6 +80,7 @@ public AccessControlManager(
    * @param plan An Execution plan
    */
   public void apply(final Topology topology, ExecutionPlan plan) throws IOException {
+    julieRoles.validateTopology(topology);
     List<Action> actions = buildProjectActions(topology);
     actions.addAll(buildPlatformLevelActions(topology));
     buildUpdateBindingsActions(actions, loadActualClusterStateIfAvailable(plan)).forEach(plan::add);
@@ -99,7 +112,7 @@ private Set<TopologyAclBinding> providerBindings() {
    * @param topology A topology file
    * @return List<Action> A list of actions required based on the parameters
    */
-  public List<Action> buildProjectActions(Topology topology) {
+  public List<Action> buildProjectActions(Topology topology) throws IOException {
     List<Action> actions = new ArrayList<>();
 
     for (Project project : topology.getProjects()) {
@@ -132,6 +145,16 @@ public List<Action> buildProjectActions(Topology topology) {
       }
 
       syncRbacRawRoles(project.getRbacRawRoles(), topicPrefix, actions);
+
+      for (Map.Entry<String, List<Other>> other : project.getOthers().entrySet()) {
+        if (julieRoles.size() == 0) {
+          throw new IOException(
+              "Custom JulieRoles are being used without providing the required config file.");
+        }
+        actions.add(
+            new BuildBindingsForRole(
+                bindingsBuilder, julieRoles.get(other.getKey()), other.getValue()));
+      }
     }
     return actions;
   }

src/main/java/com/purbon/kafka/topology/BindingsBuilderProvider.java

@@ -2,9 +2,11 @@
 
 import com.purbon.kafka.topology.exceptions.ConfigurationException;
 import com.purbon.kafka.topology.model.Component;
+import com.purbon.kafka.topology.model.JulieRoleAcl;
 import com.purbon.kafka.topology.model.users.Connector;
 import com.purbon.kafka.topology.model.users.Consumer;
 import com.purbon.kafka.topology.model.users.KSqlApp;
+import com.purbon.kafka.topology.model.users.Other;
 import com.purbon.kafka.topology.model.users.Producer;
 import com.purbon.kafka.topology.model.users.platform.KsqlServerInstance;
 import com.purbon.kafka.topology.model.users.platform.SchemaRegistryInstance;
@@ -56,4 +58,7 @@ default List<TopologyAclBinding> setClusterLevelRole(
   Collection<TopologyAclBinding> buildBindingsForKSqlServer(KsqlServerInstance ksqlServer);
 
   Collection<TopologyAclBinding> buildBindingsForKSqlApp(KSqlApp app, String prefix);
+
+  Collection<TopologyAclBinding> buildBindingsForJulieRole(
+      Other other, String name, List<JulieRoleAcl> acls) throws IOException;
 }

src/main/java/com/purbon/kafka/topology/Configuration.java

@@ -5,24 +5,32 @@
 
 import com.purbon.kafka.topology.api.ksql.KsqlClientConfig;
 import com.purbon.kafka.topology.exceptions.ConfigurationException;
+import com.purbon.kafka.topology.model.JulieRoles;
 import com.purbon.kafka.topology.model.Project;
 import com.purbon.kafka.topology.model.Topic;
 import com.purbon.kafka.topology.model.Topology;
+import com.purbon.kafka.topology.serdes.JulieRolesSerdes;
 import com.purbon.kafka.topology.serdes.TopologySerdes.FileType;
 import com.purbon.kafka.topology.utils.BasicAuth;
 import com.purbon.kafka.topology.utils.Pair;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigException;
 import com.typesafe.config.ConfigFactory;
 import java.io.File;
+import java.io.IOException;
+import java.nio.file.Paths;
 import java.util.*;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 public class Configuration {
 
+  private static final Logger LOGGER = LogManager.getLogger(Configuration.class);
+
   private final Map<String, String> cliParams;
   private final Config config;
 
@@ -361,6 +369,10 @@ public String getKafkaConnectClusterId() {
     return config.getString(MDS_KC_CLUSTER_ID_CONFIG);
   }
 
+  public String getKsqlDBClusterID() {
+    return config.getString(MDS_KSQLDB_CLUSTER_ID_CONFIG);
+  }
+
   public Optional<String> getSslTrustStoreLocation() {
     try {
       return Optional.of(config.getString(SSL_TRUSTSTORE_LOCATION));
@@ -477,4 +489,18 @@ public Optional<BasicAuth> getMdsBasicAuth() {
     }
     return Optional.ofNullable(auth);
   }
+
+  public JulieRoles getJulieRoles() throws IOException {
+    JulieRolesSerdes serdes = new JulieRolesSerdes();
+    try {
+      String path = config.getString(JULIE_ROLES);
+      return serdes.deserialise(Paths.get(path).toFile());
+    } catch (ConfigException.Missing | ConfigException.WrongType ex) {
+      LOGGER.debug(ex);
+      return new JulieRoles();
+    } catch (IOException e) {
+      LOGGER.error(e);
+      throw e;
+    }
+  }
 }

src/main/java/com/purbon/kafka/topology/Constants.java

@@ -41,6 +41,7 @@ public class Constants {
   public static final String MDS_KAFKA_CLUSTER_ID_CONFIG = "topology.builder.mds.kafka.cluster.id";
   static final String MDS_SR_CLUSTER_ID_CONFIG = "topology.builder.mds.schema.registry.cluster.id";
   static final String MDS_KC_CLUSTER_ID_CONFIG = "topology.builder.mds.kafka.connect.cluster.id";
+  static final String MDS_KSQLDB_CLUSTER_ID_CONFIG = "topology.builder.mds.ksqldb.cluster.id";
 
   public static final String CONFLUENT_SCHEMA_REGISTRY_URL_CONFIG = "schema.registry.url";
   static final String CONFLUENT_MONITORING_TOPIC_CONFIG = "confluent.monitoring.topic";
@@ -116,4 +117,6 @@ public class Constants {
   public static final String SSL_KEYSTORE_LOCATION = "ssl.keystore.location";
   public static final String SSL_KEYSTORE_PASSWORD = "ssl.keystore.password";
   public static final String SSL_KEY_PASSWORD = "ssl.key.password";
+
+  public static final String JULIE_ROLES = "julie.roles";
 }

src/main/java/com/purbon/kafka/topology/JulieOps.java

@@ -131,7 +131,8 @@ public static JulieOps build(
     config.validateWith(topology);
 
     AccessControlManager accessControlManager =
-        new AccessControlManager(accessControlProvider, bindingsBuilderProvider, config);
+        new AccessControlManager(
+            accessControlProvider, bindingsBuilderProvider, config.getJulieRoles(), config);
 
     RestService restService = new RestService(config.getConfluentSchemaRegistryUrl());
     Map<String, ?> schemaRegistryConfig = config.asMap();