Fix: RBAC Subject and Cluster level binding (#301)

* Fix: RBAC Subject and Cluster level binding

* Fix: Cluster scoped role cannot be used with resource binding api

Co-authored-by: Ludovic BOUTROS <lboutros@solocal.com>

src/main/java/com/purbon/kafka/topology/api/mds/MDSApiClient.java

@@ -3,6 +3,7 @@
 import static com.purbon.kafka.topology.api.mds.RequestScope.RESOURCE_NAME;
 import static com.purbon.kafka.topology.api.mds.RequestScope.RESOURCE_PATTERN_TYPE;
 import static com.purbon.kafka.topology.api.mds.RequestScope.RESOURCE_TYPE;
+import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.isClusterScopedRole;
 
 import com.purbon.kafka.topology.Configuration;
 import com.purbon.kafka.topology.clients.JulieHttpClient;
@@ -77,22 +78,30 @@ public TopologyAclBinding bindClusterRole(
     return binding;
   }
 
-  public void bindRequest(TopologyAclBinding binding) throws IOException {
+  private boolean isBindingWithResources(TopologyAclBinding binding) {
+    return !binding.getScope().getResources().isEmpty();
+  }
 
+  MDSRequest buildRequest(TopologyAclBinding binding) {
     String url = binding.getPrincipal() + "/roles/" + binding.getOperation();
-    if (!binding.getResourceType().equals(ResourceType.CLUSTER.name())) {
+    String jsonEntity;
+
+    if (isBindingWithResources(binding) && !isClusterScopedRole(binding.getOperation())) {
       url = url + "/bindings";
+      jsonEntity = binding.getScope().asJson();
+    } else {
+      jsonEntity = binding.getScope().clustersAsJson();
     }
+    LOGGER.debug("bind.entity: " + jsonEntity);
 
+    return new MDSRequest(url, jsonEntity);
+  }
+
+  public void bindRequest(TopologyAclBinding binding) throws IOException {
+    MDSRequest mdsRequest = buildRequest(binding);
     try {
-      String jsonEntity;
-      if (binding.getResourceType().equals(ResourceType.CLUSTER.name())) {
-        jsonEntity = binding.getScope().clustersAsJson();
-      } else {
-        jsonEntity = binding.getScope().asJson();
-      }
-      LOGGER.debug("bind.entity: " + jsonEntity);
-      doPost("/security/1.0/principals/" + url, jsonEntity);
+      LOGGER.debug("bind.entity: " + mdsRequest.getJsonEntity());
+      doPost("/security/1.0/principals/" + mdsRequest.getUrl(), mdsRequest.getJsonEntity());
     } catch (IOException e) {
       LOGGER.error(e);
       throw e;

src/main/java/com/purbon/kafka/topology/api/mds/MDSRequest.java

@@ -0,0 +1,19 @@
+package com.purbon.kafka.topology.api.mds;
+
+public class MDSRequest {
+  private final String url;
+  private final String jsonEntity;
+
+  public MDSRequest(String url, String jsonEntity) {
+    this.url = url;
+    this.jsonEntity = jsonEntity;
+  }
+
+  public String getUrl() {
+    return url;
+  }
+
+  public String getJsonEntity() {
+    return jsonEntity;
+  }
+}

src/main/java/com/purbon/kafka/topology/roles/rbac/RBACBindingsBuilder.java

@@ -366,7 +366,12 @@ public List<TopologyAclBinding> setClusterLevelRole(
   @Override
   public List<TopologyAclBinding> setSchemaAuthorization(String principal, List<String> subjects) {
     return subjects.stream()
-        .map(subject -> apiClient.bind(principal, RESOURCE_OWNER).forSchemaSubject(subject).apply())
+        .map(
+            subject ->
+                apiClient
+                    .bind(principal, RESOURCE_OWNER)
+                    .forSchemaSubject(subject)
+                    .apply("Subject", subject))
         .filter(Objects::nonNull)
         .collect(Collectors.toList());
   }

src/main/java/com/purbon/kafka/topology/roles/rbac/RBACPredefinedRoles.java

@@ -9,4 +9,10 @@ public class RBACPredefinedRoles {
 
   public static final String SECURITY_ADMIN = "SecurityAdmin";
   public static final String SYSTEM_ADMIN = "SystemAdmin";
+
+  public static boolean isClusterScopedRole(String role) {
+    return !DEVELOPER_READ.equals(role)
+            && !DEVELOPER_WRITE.equals(role)
+            && !RESOURCE_OWNER.equals(role);
+  }
 }

src/test/java/com/purbon/kafka/topology/api/mds/MDSApiClientTest.java

@@ -0,0 +1,46 @@
+package com.purbon.kafka.topology.api.mds;
+
+import static com.purbon.kafka.topology.roles.rbac.RBACPredefinedRoles.DEVELOPER_READ;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+import com.purbon.kafka.topology.roles.TopologyAclBinding;
+import org.junit.Test;
+
+public class MDSApiClientTest {
+  MDSApiClient apiClient = new MDSApiClient("http://not_used:8090");
+
+  @Test
+  public void testBindSubjectRole() {
+    String principal = "User:foo";
+    String subject = "topic-value";
+
+    TopologyAclBinding binding =
+        apiClient
+            .bind(principal, DEVELOPER_READ)
+            .forSchemaSubject(subject)
+            .apply("Subject", subject);
+
+    MDSRequest mdsRequest = apiClient.buildRequest(binding);
+
+    assertThat(mdsRequest.getUrl()).isEqualTo("User:foo/roles/DeveloperRead/bindings");
+    assertThat(mdsRequest.getJsonEntity())
+        .isEqualTo(
+            "{\"resourcePatterns\":[{\"name\":\"Subject:topic-value\",\"patternType\":\"LITERAL\",\"resourceType\":\"Subject\"}],\"scope\":{\"clusters\":{\"kafka-cluster\":\"\",\"schema-registry-cluster\":\"\"}}}");
+  }
+
+  @Test
+  public void testBindSubjectRoleWithoutResourceType() {
+    String principal = "User:foo";
+    String subject = "topic-value";
+
+    TopologyAclBinding binding =
+        apiClient.bind(principal, DEVELOPER_READ).forSchemaSubject(subject).apply();
+
+    MDSRequest mdsRequest = apiClient.buildRequest(binding);
+
+    assertThat(mdsRequest.getUrl()).isEqualTo("User:foo/roles/DeveloperRead/bindings");
+    assertThat(mdsRequest.getJsonEntity())
+        .isEqualTo(
+            "{\"resourcePatterns\":[{\"name\":\"Subject:topic-value\",\"patternType\":\"LITERAL\",\"resourceType\":\"Subject\"}],\"scope\":{\"clusters\":{\"kafka-cluster\":\"\",\"schema-registry-cluster\":\"\"}}}");
+  }
+}